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

from io import StringIO
import logging
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest import mock

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

from testing_support import coverage_utils  # noqa: E402
import gclient_utils  # noqa: E402
import git_cache  # noqa: E402
import git_common  # noqa: E402


class GitCacheTest(unittest.TestCase):
    def setUp(self):
        self.cache_dir = tempfile.mkdtemp(prefix="git_cache_test_")
        self.addCleanup(shutil.rmtree, self.cache_dir, ignore_errors=True)
        self.origin_dir = tempfile.mkdtemp(suffix="origin.git")
        self.addCleanup(shutil.rmtree, self.origin_dir, ignore_errors=True)
        git_cache.Mirror.SetCachePath(self.cache_dir)

        # Ensure git_cache works with safe.bareRepository.
        mock.patch.dict(
            "os.environ",
            {
                "GIT_CONFIG_GLOBAL": os.path.join(self.cache_dir, ".gitconfig"),
            },
        ).start()
        self.addCleanup(mock.patch.stopall)
        self.git(
            [
                "config",
                "--file",
                os.path.join(self.cache_dir, ".gitconfig"),
                "--add",
                "safe.bareRepository",
                "explicit",
            ]
        )

    def git(self, cmd, cwd=None):
        cwd = cwd or self.origin_dir
        git = "git.bat" if sys.platform == "win32" else "git"
        subprocess.check_call([git] + cmd, cwd=cwd)

    def _cacheHasTag(self, cache_dir: str, tag: str = "TAG") -> bool:
        git = "git.bat" if sys.platform == "win32" else "git"
        return (
            subprocess.call(
                [
                    git,
                    "--git-dir",
                    cache_dir,
                    "show-ref",
                    "--verify",
                    "--quiet",
                    f"refs/tags/{tag}",
                ]
            )
            == 0
        )

    def _supportsReftable(self) -> bool:
        return git_common.meets_git_version((2, 45, 0))

    def _setDefaultRefFormat(self, ref_format: str):
        """Makes the host git default to ref_format for new/re-inits."""
        self.git(
            [
                "config",
                "--file",
                os.path.join(self.cache_dir, ".gitconfig"),
                "init.defaultRefFormat",
                ref_format,
            ]
        )

    def _makeBareMirrorWithHead(
        self, head_ref: str, ref_format: str = "reftable"
    ) -> git_cache.Mirror:
        """Creates the mirror as a bare repo with HEAD pointing at head_ref."""
        mirror = git_cache.Mirror(self.origin_dir)
        gclient_utils.rmtree(mirror.mirror_path)
        cmd = ["init", "-q", "--bare"]
        if ref_format:
            cmd.append(f"--ref-format={ref_format}")
        cmd.append(mirror.mirror_path)
        self.git(cmd)
        self.git(
            ["--git-dir", mirror.mirror_path, "symbolic-ref", "HEAD", head_ref]
        )
        return mirror

    @mock.patch("git_cache.Mirror.bootstrap_repo", return_value=True)
    def testEnsureBootstrappedHandlesStaleMasterHead(self, _mock_bootstrap):
        # HEAD is read via git (symbolic-ref), so a cache left pointing at
        # refs/heads/master (with no such ref) is detected as stale and nuked,
        # while a healthy refs/heads/main cache is kept -- regardless of the
        # on-disk ref backend (loose, packed, or reftable).
        ref_formats = ["files"]
        if self._supportsReftable():
            ref_formats.append("reftable")
        cases = [
            # (head_ref, expect_deleted)
            ("refs/heads/master", True),
            ("refs/heads/main", False),
        ]
        for ref_format in ref_formats:
            for head_ref, expect_deleted in cases:
                with self.subTest(head_ref=head_ref, ref_format=ref_format):
                    mirror = self._makeBareMirrorWithHead(
                        head_ref, ref_format=ref_format
                    )
                    with mock.patch(
                        "git_cache.gclient_utils.rmtree"
                    ) as mock_rmtree:
                        mirror._ensure_bootstrapped(None, True, False)
                    if expect_deleted:
                        mock_rmtree.assert_any_call(mirror.mirror_path)
                    else:
                        mock_rmtree.assert_not_called()

    def testParseFetchSpec(self):
        testData = [
            ([], []),
            (
                ["main"],
                [("+refs/heads/main:refs/heads/main", r"\+refs/heads/main:.*")],
            ),
            (
                ["main/"],
                [("+refs/heads/main:refs/heads/main", r"\+refs/heads/main:.*")],
            ),
            (
                ["+main"],
                [("+refs/heads/main:refs/heads/main", r"\+refs/heads/main:.*")],
            ),
            (
                ["master"],
                [
                    (
                        "+refs/heads/master:refs/heads/master",
                        r"\+refs/heads/master:.*",
                    )
                ],
            ),
            (
                ["master/"],
                [
                    (
                        "+refs/heads/master:refs/heads/master",
                        r"\+refs/heads/master:.*",
                    )
                ],
            ),
            (
                ["+master"],
                [
                    (
                        "+refs/heads/master:refs/heads/master",
                        r"\+refs/heads/master:.*",
                    )
                ],
            ),
            (
                ["refs/heads/*"],
                [("+refs/heads/*:refs/heads/*", r"\+refs/heads/\*:.*")],
            ),
            (
                ["foo/bar/*", "baz"],
                [
                    (
                        "+refs/heads/foo/bar/*:refs/heads/foo/bar/*",
                        r"\+refs/heads/foo/bar/\*:.*",
                    ),
                    ("+refs/heads/baz:refs/heads/baz", r"\+refs/heads/baz:.*"),
                ],
            ),
            (
                ["refs/foo/*:refs/bar/*"],
                [("+refs/foo/*:refs/bar/*", r"\+refs/foo/\*:.*")],
            ),
        ]

        mirror = git_cache.Mirror("test://phony.example.biz")
        for fetch_specs, expected in testData:
            mirror = git_cache.Mirror(
                "test://phony.example.biz", refs=fetch_specs
            )
            self.assertEqual(mirror.fetch_specs, set(expected))

    def testPopulate(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate()

    def testPopulateResetFetchConfig(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate()

        # Add a bad refspec to the cache's fetch config.
        cache_dir = os.path.join(
            self.cache_dir, mirror.UrlToCacheDir(self.origin_dir)
        )
        self.git(
            [
                "--git-dir",
                cache_dir,
                "config",
                "--add",
                "remote.origin.fetch",
                "+refs/heads/foo:refs/heads/foo",
            ],
            cwd=cache_dir,
        )

        mirror.populate(reset_fetch_config=True)

    def testPopulateTwice(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate()

        mirror.populate()

    def testPopulateDeletesTmpPackFiles(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate()

        # Create tmp pack files.
        pack_dir = os.path.join(mirror.mirror_path, "objects", "pack")
        os.makedirs(pack_dir, exist_ok=True)
        tmp_pack_path = os.path.join(pack_dir, "tmp_pack_abc")
        tmp_idx_path = os.path.join(pack_dir, ".tmp-1234-pack-def.idx")
        with open(tmp_pack_path, "w") as f:
            f.write("content")
        with open(tmp_idx_path, "w") as f:
            f.write("idx content")

        self.assertTrue(os.path.exists(tmp_pack_path))
        self.assertTrue(os.path.exists(tmp_idx_path))

        mirror.populate()

        # The temporary files should be deleted.
        self.assertFalse(os.path.exists(tmp_pack_path))
        self.assertFalse(os.path.exists(tmp_idx_path))

    @mock.patch("sys.stdout", StringIO())
    def testPruneRequired(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["checkout", "-b", "foo"])
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )
        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate()
        self.git(["checkout", "-b", "foo_tmp", "foo"])
        self.git(["branch", "-D", "foo"])
        self.git(["checkout", "-b", "foo/bar", "foo_tmp"])
        mirror.populate()
        self.assertNotIn(
            git_cache.GIT_CACHE_CORRUPT_MESSAGE, sys.stdout.getvalue()
        )

    @mock.patch("sys.stdout", StringIO())
    def testBadInit(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

        mirror = git_cache.Mirror(self.origin_dir)

        # Simulate init being interrupted during fetch phase.
        with mock.patch.object(mirror, "_fetch"):
            mirror.populate()

        # Corrupt message is not expected at this point since it was
        # "interrupted".
        self.assertNotIn(
            git_cache.GIT_CACHE_CORRUPT_MESSAGE, sys.stdout.getvalue()
        )

        # We call mirror.populate() without _fetch patched. This time, a
        # sentient file should prompt cache deletion.
        mirror.populate()
        self.assertIn(
            git_cache.GIT_CACHE_CORRUPT_MESSAGE, sys.stdout.getvalue()
        )

    def testPopulateInitObjectFormat(self):
        """A fresh mirror pins the bare repo to the remote object format.

        An existing mirror that is re-initialized must not pass
        --object-format, since the object format cannot be changed after
        creation.
        """
        self._makeGitRepo()
        mirror = git_cache.Mirror(self.origin_dir)
        real_run_git = mirror.RunGit

        def populate_capturing_init():
            init_calls = []

            def record(cmd, *args, **kwargs):
                if cmd[:1] == ["init"]:
                    init_calls.append(list(cmd))
                return real_run_git(cmd, *args, **kwargs)

            with mock.patch.object(mirror, "RunGit", side_effect=record):
                mirror.populate()
            return init_calls

        with mock.patch.object(
            git_cache.scm.GIT,
            "GetRemoteObjectFormat",
            return_value="sha1",
        ) as mock_object_format:
            fresh_init_calls = populate_capturing_init()
            existing_init_calls = populate_capturing_init()

        mock_object_format.assert_called_with(self.origin_dir)

        with self.subTest("fresh mirror pins object format"):
            self.assertTrue(
                any("--object-format=sha1" in cmd for cmd in fresh_init_calls),
                fresh_init_calls,
            )
            if self._supportsReftable():
                self.assertTrue(
                    any(
                        "--ref-format=files" in cmd for cmd in fresh_init_calls
                    ),
                    fresh_init_calls,
                )

        with self.subTest("existing mirror omits object format"):
            self.assertTrue(existing_init_calls)
            for cmd in existing_init_calls:
                self.assertNotIn("--object-format=sha1", cmd)
                if self._supportsReftable():
                    self.assertIn("--ref-format=files", cmd)

    def testBootstrapRepoPinsRefFormatFiles(self):
        mirror = git_cache.Mirror("https://chromium.googlesource.com/foo/bar")
        init_calls = []
        real_run_git = mirror.RunGit

        def record_git(cmd, *args, **kwargs):
            if cmd[:1] == ["init"]:
                init_calls.append(list(cmd))
            return real_run_git(cmd, *args, **kwargs)

        with (
            mock.patch.object(
                git_cache.Gsutil,
                "check_call",
                return_value=(
                    0,
                    "gs://chromium-git-cache/v2/foo-bar/100.ready\n"
                    "gs://chromium-git-cache/v2/foo-bar/100/\n",
                    "",
                ),
            ),
            mock.patch.object(git_cache.Gsutil, "call", return_value=0),
            mock.patch.object(mirror, "RunGit", side_effect=record_git),
        ):
            target_dir = os.path.join(self.cache_dir, "test_target")
            self.assertTrue(mirror.bootstrap_repo(target_dir))

        if self._supportsReftable():
            self.assertTrue(
                any("--ref-format=files" in cmd for cmd in init_calls),
                init_calls,
            )

    def testEnsureBootstrappedPreservesExistingRefFormat(self):
        """Re-initializing an existing mirror must keep its ref backend.

        git rejects an init that would change the ref storage format, so
        pinning files unconditionally would permanently break mirrors created
        before the files pin landed -- e.g. by an older depot_tools on a host
        whose git defaults to reftable.
        """
        if not self._supportsReftable():
            self.skipTest("git is too old for --ref-format")

        mirror = self._makeBareMirrorWithHead(
            "refs/heads/main", ref_format="reftable"
        )
        self.assertEqual("reftable", mirror._get_ref_format())

        with (
            mock.patch.object(
                git_cache.Mirror, "bootstrap_repo", return_value=False
            ),
            mock.patch.object(git_cache.Mirror, "_set_symbolic_ref"),
        ):
            mirror._ensure_bootstrapped(None, True, False)

        self.assertEqual("reftable", mirror._get_ref_format())

    def testEnsureBootstrappedFilesMirrorOnReftableHost(self):
        """A files mirror must survive re-init where git defaults to reftable.

        This is the reported breakage: plain `git init --bare` picks the host
        default, then dies on the existing refs/heads directory.
        """
        if not self._supportsReftable():
            self.skipTest("git is too old for --ref-format")

        mirror = self._makeBareMirrorWithHead(
            "refs/heads/main", ref_format="files"
        )
        self._setDefaultRefFormat("reftable")

        with (
            mock.patch.object(
                git_cache.Mirror, "bootstrap_repo", return_value=False
            ),
            mock.patch.object(git_cache.Mirror, "_set_symbolic_ref"),
        ):
            mirror._ensure_bootstrapped(None, True, False)

        self.assertEqual("files", mirror._get_ref_format())

    def testEnsureBootstrappedRepairsPoisonedRefFormat(self):
        """A failed reftable init leaves a config that hides every ref.

        git writes extensions.refStorage before it fails on refs/heads, so a
        mirror broken by a pre-fix depot_tools reports no refs and re-fails
        every init. The refs are intact; re-init must repair rather than
        inherit the bad format.
        """
        if not self._supportsReftable():
            self.skipTest("git is too old for --ref-format")

        self._makeGitRepoWithTag()
        mirror = git_cache.Mirror(self.origin_dir)
        gclient_utils.rmtree(mirror.mirror_path)
        self.git(
            [
                "clone",
                "--bare",
                "--ref-format=files",
                self.origin_dir,
                mirror.mirror_path,
            ]
        )
        self.assertTrue(self._cacheHasTag(mirror.mirror_path))

        # Poison it exactly the way a pre-fix depot_tools did.
        self._setDefaultRefFormat("reftable")
        with self.assertRaises(subprocess.CalledProcessError):
            self.git(["--git-dir", mirror.mirror_path, "init", "--bare"])
        self.assertEqual("reftable", mirror._get_ref_format())
        self.assertFalse(self._cacheHasTag(mirror.mirror_path))

        with (
            mock.patch.object(
                git_cache.Mirror, "bootstrap_repo", return_value=False
            ),
            mock.patch.object(git_cache.Mirror, "_set_symbolic_ref"),
        ):
            mirror._ensure_bootstrapped(None, True, False)

        self.assertEqual("files", mirror._get_ref_format())
        self.assertTrue(self._cacheHasTag(mirror.mirror_path))

    def testRefFormatArgsSkippedOnOldGit(self):
        """The --ref-format flag only exists in git 2.45+."""
        with mock.patch.object(
            git_cache.git_common, "meets_git_version", return_value=False
        ):
            self.assertEqual([], git_cache._ref_format_args())
        with mock.patch.object(
            git_cache.git_common, "meets_git_version", return_value=True
        ):
            self.assertEqual(
                ["--ref-format=files"], git_cache._ref_format_args()
            )
            self.assertEqual(
                ["--ref-format=reftable"],
                git_cache._ref_format_args("reftable"),
            )

    def _makeGitRepo(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

    def _makeGitRepoWithTag(self):
        self._makeGitRepo()
        self.git(["tag", "TAG"])
        self.git(["pack-refs"])

    def testPopulateFetchTagsByDefault(self):
        self._makeGitRepoWithTag()

        # Default behaviour includes tags.
        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate()

        cache_dir = os.path.join(
            self.cache_dir, mirror.UrlToCacheDir(self.origin_dir)
        )
        self.assertTrue(self._cacheHasTag(cache_dir))

    def testPopulateFetchWithoutTags(self):
        self._makeGitRepoWithTag()

        # Ask to not include tags.
        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate(no_fetch_tags=True)

        cache_dir = os.path.join(
            self.cache_dir, mirror.UrlToCacheDir(self.origin_dir)
        )
        self.assertFalse(self._cacheHasTag(cache_dir))

    def testPopulateResetFetchConfigEmptyFetchConfig(self):
        self.git(["init", "-q"])
        with open(os.path.join(self.origin_dir, "foo"), "w") as f:
            f.write("touched\n")
        self.git(["add", "foo"])
        self.git(
            [
                "-c",
                "user.name=Test user",
                "-c",
                "user.email=joj@test.com",
                "commit",
                "-m",
                "foo",
            ]
        )

        mirror = git_cache.Mirror(self.origin_dir)
        mirror.populate(reset_fetch_config=True)

    @mock.patch("gclient_utils.exponential_backoff_retry")
    def testSetSymbolicRefIgnoreUnknown(self, mock_retry):
        mock_retry.return_value = "HEAD branch: (unknown)"
        mirror = git_cache.Mirror(self.origin_dir)
        with mock.patch.object(mirror, "RunGit") as mock_rungit:
            mirror._set_symbolic_ref()
            mock_rungit.assert_not_called()

    @mock.patch("gclient_utils.exponential_backoff_retry")
    def testSetSymbolicRefKnown(self, mock_retry):
        mock_retry.return_value = "HEAD branch: main"
        mirror = git_cache.Mirror(self.origin_dir)
        with mock.patch.object(mirror, "RunGit") as mock_rungit:
            mirror._set_symbolic_ref()
            mock_rungit.assert_called_once_with(
                ["symbolic-ref", "HEAD", "refs/heads/main"]
            )


class GitCacheDirTest(unittest.TestCase):
    def setUp(self):
        try:
            delattr(git_cache.Mirror, "cachepath")
        except AttributeError:
            pass
        super(GitCacheDirTest, self).setUp()

    def tearDown(self):
        try:
            delattr(git_cache.Mirror, "cachepath")
        except AttributeError:
            pass
        super(GitCacheDirTest, self).tearDown()

    def test_git_config_read(self):
        (fd, tmpFile) = tempfile.mkstemp()
        old = git_cache.Mirror._GIT_CONFIG_LOCATION
        try:
            try:
                os.write(fd, b'[cache]\n  cachepath="hello world"\n')
            finally:
                os.close(fd)

            git_cache.Mirror._GIT_CONFIG_LOCATION = ["-f", tmpFile]

            self.assertEqual(git_cache.Mirror.GetCachePath(), "hello world")
        finally:
            git_cache.Mirror._GIT_CONFIG_LOCATION = old
            os.remove(tmpFile)

    def test_environ_read(self):
        path = os.environ.get("GIT_CACHE_PATH")
        config = os.environ.get("GIT_CONFIG")
        try:
            os.environ["GIT_CACHE_PATH"] = "hello world"
            os.environ["GIT_CONFIG"] = "disabled"

            self.assertEqual(git_cache.Mirror.GetCachePath(), "hello world")
        finally:
            for name, val in zip(
                ("GIT_CACHE_PATH", "GIT_CONFIG"), (path, config)
            ):
                if val is None:
                    os.environ.pop(name, None)
                else:
                    os.environ[name] = val

    def test_manual_set(self):
        git_cache.Mirror.SetCachePath("hello world")
        self.assertEqual(git_cache.Mirror.GetCachePath(), "hello world")

    def test_unconfigured(self):
        path = os.environ.get("GIT_CACHE_PATH")
        config = os.environ.get("GIT_CONFIG")
        try:
            os.environ.pop("GIT_CACHE_PATH", None)
            os.environ["GIT_CONFIG"] = "disabled"

            with self.assertRaisesRegex(RuntimeError, r"cache\.cachepath"):
                git_cache.Mirror.GetCachePath()

            # negatively cached value still raises
            with self.assertRaisesRegex(RuntimeError, r"cache\.cachepath"):
                git_cache.Mirror.GetCachePath()
        finally:
            for name, val in zip(
                ("GIT_CACHE_PATH", "GIT_CONFIG"), (path, config)
            ):
                if val is None:
                    os.environ.pop(name, None)
                else:
                    os.environ[name] = val


class BootstrapConcurrencyTest(unittest.TestCase):
    def test_default_tracks_sync_parallelism(self):
        with mock.patch.dict("os.environ", clear=False):
            os.environ.pop("GIT_CACHE_BOOTSTRAP_CONCURRENCY", None)
            self.assertEqual(
                git_cache._bootstrap_concurrency(),
                max(8, gclient_utils.NumLocalCpus()),
            )

    def test_override(self):
        with mock.patch.dict(
            "os.environ", {"GIT_CACHE_BOOTSTRAP_CONCURRENCY": "32"}
        ):
            self.assertEqual(git_cache._bootstrap_concurrency(), 32)

    def test_clamped_to_at_least_one(self):
        with mock.patch.dict(
            "os.environ", {"GIT_CACHE_BOOTSTRAP_CONCURRENCY": "0"}
        ):
            self.assertEqual(git_cache._bootstrap_concurrency(), 1)

    def test_garbage_falls_back_to_default(self):
        with mock.patch.dict(
            "os.environ", {"GIT_CACHE_BOOTSTRAP_CONCURRENCY": "lots"}
        ):
            self.assertEqual(
                git_cache._bootstrap_concurrency(),
                max(8, gclient_utils.NumLocalCpus()),
            )


class GetBootstrapDefaultBranchTest(unittest.TestCase):
    def setUp(self):
        self.cache_dir = tempfile.mkdtemp(prefix="gc_defbranch_")
        self.addCleanup(shutil.rmtree, self.cache_dir, ignore_errors=True)
        git_cache.Mirror.SetCachePath(self.cache_dir)

    def _patched(self, ls_lines, head="ref: refs/heads/main\n", ls_code=0):
        class _FakeGsutil:
            def __init__(self, *a, **k):
                pass

            def check_call(self, *args):
                if args[0] == "ls":
                    return (ls_code, "\n".join(ls_lines), "")
                if args[0] == "cat":
                    return (0, head, "")
                return (1, "", "")

        return mock.patch.object(git_cache, "Gsutil", _FakeGsutil)

    def test_reads_default_branch_from_snapshot_head(self):
        m = git_cache.Mirror("https://chromium.googlesource.com/foo/bar")
        gp = m._gs_path
        with self._patched(["%s/42/" % gp, "%s/42.ready" % gp]):
            self.assertEqual(m.get_bootstrap_default_branch(), "main")

    def test_respects_non_main_default(self):
        m = git_cache.Mirror("https://chromium.googlesource.com/foo/bar")
        gp = m._gs_path
        with self._patched(
            ["%s/7/" % gp, "%s/7.ready" % gp], head="ref: refs/heads/master\n"
        ):
            self.assertEqual(m.get_bootstrap_default_branch(), "master")

    def test_bucket_without_snapshot_returns_none(self):
        # Supported host, but no snapshot uploaded for this repo yet.
        m = git_cache.Mirror("https://chromium.googlesource.com/foo/bar")
        with self._patched([]):
            self.assertIsNone(m.get_bootstrap_default_branch())

    def test_unsupported_host_returns_none(self):
        # Host has no bootstrap bucket at all.
        with mock.patch.dict("os.environ", clear=False):
            os.environ.pop("OVERRIDE_BOOTSTRAP_BUCKET", None)
            m = git_cache.Mirror("https://unknown.example.com/foo/bar")
            self.assertIsNone(m.get_bootstrap_default_branch())


class BootstrapBucketTest(unittest.TestCase):
    def setUp(self):
        self.cache_dir = tempfile.mkdtemp(prefix="gc_bucket_")
        self.addCleanup(shutil.rmtree, self.cache_dir, ignore_errors=True)
        git_cache.Mirror.SetCachePath(self.cache_dir)
        mock.patch.dict("os.environ", clear=False).start()
        self.addCleanup(mock.patch.stopall)
        os.environ.pop("OVERRIDE_BOOTSTRAP_BUCKET", None)

    def test_chromium_host(self):
        m = git_cache.Mirror("https://chromium.googlesource.com/v8/v8")
        self.assertEqual(m.bootstrap_bucket, "chromium-git-cache")

    def test_additional_public_hosts(self):
        for host in (
            "dawn",
            "skia",
            "webrtc",
            "pdfium",
            "boringssl",
            "aomedia",
            "quiche",
            "swiftshader",
            "android",
        ):
            m = git_cache.Mirror("https://%s.googlesource.com/x" % host)
            self.assertEqual(m.bootstrap_bucket, "chromium-git-cache", host)

    def test_unknown_host_returns_none(self):
        self.assertIsNone(
            git_cache.Mirror("https://example.com/x").bootstrap_bucket
        )

    def test_aliased_url(self):
        url = (
            "https://chrome-internal.googlesource.com/"
            "chrome/experimental/chromium/src"
        )
        m = git_cache.Mirror(url)
        self.assertEqual(m.basedir, "chromium.googlesource.com-chromium-src")
        self.assertEqual(m.bootstrap_bucket, "chromium-git-cache")
        expected_gs_path = (
            "gs://chromium-git-cache/v2/chromium.googlesource.com-chromium-src"
        )
        self.assertEqual(m._gs_path, expected_gs_path)

    def test_override_env_wins(self):
        with mock.patch.dict(
            "os.environ", {"OVERRIDE_BOOTSTRAP_BUCKET": "my-bucket"}
        ):
            m = git_cache.Mirror("https://example.com/x")
            self.assertEqual(m.bootstrap_bucket, "my-bucket")

    def test_supported_project_only_comprehensive_hosts(self):
        # Only fully-snapshotted hosts get gc.autopacklimit=0 / re-bootstrap.
        self.assertTrue(
            git_cache.Mirror(
                "https://chromium.googlesource.com/x"
            ).supported_project()
        )
        # The other bucket hosts still seed from a snapshot when available, but
        # keep normal git pack maintenance (a repo there may 404).
        for host in (
            "dawn",
            "skia",
            "webrtc",
            "pdfium",
            "boringssl",
            "aomedia",
            "quiche",
            "swiftshader",
            "android",
        ):
            m = git_cache.Mirror("https://%s.googlesource.com/x" % host)
            self.assertFalse(m.supported_project(), host)
            self.assertEqual(m.bootstrap_bucket, "chromium-git-cache", host)

    def test_supported_project_chrome_internal(self):
        m = git_cache.Mirror("https://chrome-internal.googlesource.com/x")
        self.assertTrue(m.supported_project())

    def test_supported_project_unknown_host(self):
        m = git_cache.Mirror("https://example.com/x")
        self.assertFalse(m.supported_project())


class MirrorTest(unittest.TestCase):
    def test_same_cache_for_authenticated_and_unauthenticated_urls(self):
        # GoB can fetch a repo via two different URLs; if the url contains '/a/'
        # it forces authenticated access instead of allowing anonymous access,
        # even in the case where a repo is public. We want this in order to make
        # sure bots are authenticated and get the right quotas. However, we
        # only want to maintain a single cache for the repo.
        self.assertEqual(
            git_cache.Mirror.UrlToCacheDir(
                "https://chromium.googlesource.com/a/chromium/src.git"
            ),
            "chromium.googlesource.com-chromium-src",
        )

    def test_ssh_url_in_UrlToCacheDir_and_CacheDirToUrl(self):
        ssh_url = "git@github.com:chromium/chromium.git"
        self.assertEqual(
            git_cache.Mirror.UrlToCacheDir(ssh_url),
            "git@github.com__chromium-chromium",
        )
        self.assertEqual(
            git_cache.Mirror.CacheDirToUrl(
                git_cache.Mirror.UrlToCacheDir(ssh_url)
            ),
            ssh_url[:-4],
        )


class ObjectFormatBootstrapTest(unittest.TestCase):
    def setUp(self):
        self.cache_dir = tempfile.mkdtemp(prefix="gc_obj_fmt_")
        self.addCleanup(shutil.rmtree, self.cache_dir, ignore_errors=True)
        git_cache.Mirror.SetCachePath(self.cache_dir)

    def _init_bare_repo(self, path: str, object_format: str) -> None:
        os.makedirs(path, exist_ok=True)
        git = "git.bat" if sys.platform == "win32" else "git"
        subprocess.check_call(
            [git, "init", f"--object-format={object_format}", "--bare"],
            cwd=path,
        )

    def testBootstrapSnapshotFormatMismatch(self):
        cases = [
            ("sha1", "sha256", True),
            ("sha256", "sha1", True),
            ("sha1", "sha1", False),
            ("sha256", "sha256", False),
        ]
        for snapshot_format, remote_format, expected_discard in cases:
            label = f"snapshot={snapshot_format} remote={remote_format}"
            with self.subTest(label):
                mirror = git_cache.Mirror(
                    "https://chromium.googlesource.com/foo/bar"
                )
                gclient_utils.rmtree(mirror.mirror_path)

                def fake_bootstrap(directory: str) -> bool:
                    self._init_bare_repo(directory, snapshot_format)
                    with open(
                        os.path.join(directory, "snapshot_marker"), "w"
                    ) as f:
                        f.write("snapshot")
                    return True

                with (
                    mock.patch.object(
                        mirror, "bootstrap_repo", side_effect=fake_bootstrap
                    ),
                    mock.patch(
                        "scm.GIT.GetRemoteObjectFormat",
                        return_value=remote_format,
                    ),
                    mock.patch.object(mirror, "_set_symbolic_ref"),
                ):
                    mirror._ensure_bootstrapped(
                        depth=None, bootstrap=True, reset_fetch_config=False
                    )

                self.assertEqual(mirror._get_object_format(), remote_format)
                marker_exists = os.path.exists(
                    os.path.join(mirror.mirror_path, "snapshot_marker")
                )
                if expected_discard:
                    self.assertFalse(marker_exists)
                else:
                    self.assertTrue(marker_exists)


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.DEBUG if "-v" in sys.argv else logging.ERROR
    )
    sys.exit(
        coverage_utils.covered_main(
            (os.path.join(DEPOT_TOOLS_ROOT, "git_cache.py")),
            required_percentage=0,
        )
    )