# 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 the gs.py module."""

import contextlib
import datetime
import functools
import numbers
import os
import string
import sys
from unittest import mock

from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import gs
from chromite.lib import osutils
from chromite.lib import partial_mock
from chromite.lib import retry_stats


GS_PACKAGES_PATH = "gs://test/Packages"
GS_PACKAGES_WRONG_PATH = "gs://test/Pack"

STAT_DATE_FORMAT = "%a, %d %b %Y %H:%M:%S %Z"
NOW_RAW = datetime.datetime.now()
NOW_STRING = datetime.datetime.strftime(
    NOW_RAW.replace(tzinfo=datetime.timezone.utc), STAT_DATE_FORMAT
)

# Need to re-convert the date string into a datetime object since timezone
# isn't included in NOW_RAW from datetime.now().
NOW_DATETIME = datetime.datetime.strptime(NOW_STRING, STAT_DATE_FORMAT)
STAT_OUTPUT_NOW = f"""{GS_PACKAGES_PATH}:
    Creation time:    {NOW_STRING}
    Content-Language: en
    Content-Length:   74
    Content-Type:   application/octet-stream
    Hash (crc32c):    BBPMPA==
    Hash (md5):   ms+qSYvgI9SjXn8tW/5UpQ==
    ETag:     CNCgocbmqMACEAE=
    Generation:   1408776800850000
    Metageneration:   1
    """
STAT_OUTPUT_ERR = f"No URLs matched: {GS_PACKAGES_WRONG_PATH}"


def PatchGS(*args, **kwargs):
    """Convenience method for patching GSContext."""
    return mock.patch.object(gs.GSContext, *args, **kwargs)


class GSContextMock(partial_mock.PartialCmdMock):
    """Used to mock out the GSContext class."""

    TARGET = "chromite.lib.gs.GSContext"
    ATTRS = (
        "InitializeCache",
        "DoCommand",
        "_CRCMOD_METHOD",
        "DEFAULT_SLEEP_TIME",
        "DEFAULT_RETRIES",
        "DEFAULT_BOTO_FILE",
        "_DEFAULT_GSUTIL_BIN",
        "_DEFAULT_GSUTIL_BUILDER_BIN",
        "GSUTIL_URL",
    )
    DEFAULT_ATTR = "DoCommand"

    GSResponsePreconditionFailed = """\
Copying file:///dev/null [Content-Type=application/octet-stream]...
Uploading   gs://chromeos-throw-away-bucket/vapier/null:         0 B    \r\
Uploading   gs://chromeos-throw-away-bucket/vapier/null:         0 B    \r\
PreconditionException: 412 Precondition Failed"""

    DEFAULT_SLEEP_TIME = 0
    DEFAULT_RETRIES = 2
    TMP_ROOT = "/tmp/cros_unittest"
    DEFAULT_BOTO_FILE = "%s/boto_file" % TMP_ROOT
    _DEFAULT_GSUTIL_BIN = "%s/gsutil_bin" % TMP_ROOT
    _DEFAULT_GSUTIL_BUILDER_BIN = _DEFAULT_GSUTIL_BIN
    _CRCMOD_METHOD = "missing"
    GSUTIL_URL = None

    def __init__(self) -> None:
        partial_mock.PartialCmdMock.__init__(self, create_tempdir=True)
        self.raw_gs_cmds = []

    def _SetGSUtilUrl(self) -> None:
        tempfile = os.path.join(self.tempdir, "tempfile")
        osutils.WriteFile(tempfile, "some content")
        gsutil_path = os.path.join(self.tempdir, gs.GSContext.GSUTIL_TAR)
        cros_build_lib.CreateTarball(
            gsutil_path,
            self.tempdir,
            inputs=[os.path.basename(tempfile)],
            compression=cros_build_lib.CompressionType.NONE,
        )
        self.GSUTIL_URL = "file://%s" % gsutil_path

    def PreStart(self) -> None:
        os.environ.pop("BOTO_CONFIG", None)
        # Set it here for now, instead of mocking out Cached() directly because
        # python-mock has a bug with mocking out class methods with
        # autospec=True.
        # TODO(rcui): Change this when this is fixed in PartialMock.
        self._SetGSUtilUrl()

    def InitializeCache(self, *_args, **_kwargs) -> None:
        self._DEFAULT_GSUTIL_BIN = "gsutil"

    def DoCommand(self, inst, gsutil_cmd, **kwargs):
        result = self._results["DoCommand"].LookupResult(
            (gsutil_cmd,), hook_args=(inst, gsutil_cmd), hook_kwargs=kwargs
        )

        rc_mock = cros_test_lib.RunCommandMock()
        rc_mock.AddCmdResult(
            partial_mock.ListRegex("gsutil"),
            result.returncode,
            stdout=result.stdout,
            stderr=result.stderr,
        )

        with rc_mock:
            try:
                return self.backup["DoCommand"](inst, gsutil_cmd, **kwargs)
            finally:
                self.raw_gs_cmds.extend(
                    args[0] for args, _ in rc_mock.call_args_list
                )


class AbstractGSContextTest(cros_test_lib.MockTempDirTestCase):
    """Base class for GSContext tests."""

    def setUp(self) -> None:
        self.gs_mock = self.StartPatcher(GSContextMock())
        self.gs_mock.SetDefaultCmdResult()
        self.ctx = gs.GSContext()


class VersionTest(AbstractGSContextTest):
    """Tests GSContext.gsutil_version functionality."""

    LOCAL_PATH = "/tmp/file"
    GIVEN_REMOTE = EXPECTED_REMOTE = "gs://test/path/file"

    def testGetVersionStdout(self) -> None:
        """Simple gsutil_version fetch test from stdout."""
        self.gs_mock.AddCmdResult(
            partial_mock.In("version"),
            returncode=0,
            stdout="gsutil version 3.35\n",
        )
        self.assertEqual("3.35", self.ctx.gsutil_version)

    def testGetVersionStderr(self) -> None:
        """Simple gsutil_version fetch test from stderr."""
        self.gs_mock.AddCmdResult(
            partial_mock.In("version"),
            returncode=0,
            stderr="gsutil version 3.36\n",
        )
        self.assertEqual("3.36", self.ctx.gsutil_version)

    def testGetVersionCached(self) -> None:
        """Simple gsutil_version fetch test from cache."""
        # pylint: disable=protected-access
        self.ctx._gsutil_version = "3.37"
        self.assertEqual("3.37", self.ctx.gsutil_version)

    def testGetVersionNewFormat(self) -> None:
        """Simple gsutil_version fetch test for new gsutil output format."""
        self.gs_mock.AddCmdResult(
            partial_mock.In("version"),
            returncode=0,
            stdout="gsutil version: 4.5\n",
        )
        self.assertEqual("4.5", self.ctx.gsutil_version)

    def testGetVersionBadOutput(self) -> None:
        """Simple gsutil_version fetch test from cache."""
        self.gs_mock.AddCmdResult(
            partial_mock.In("version"), returncode=0, stdout="gobblety gook\n"
        )
        self.assertRaises(
            gs.GSContextException, getattr, self.ctx, "gsutil_version"
        )


class GetSizeTest(AbstractGSContextTest):
    """Tests GetSize functionality."""

    GETSIZE_PATH = "gs://abc/1"

    def _GetSize(self, ctx, path, **kwargs):
        return ctx.GetSize(path, **kwargs)

    def GetSize(self, ctx=None, **kwargs):
        if ctx is None:
            ctx = self.ctx
        return self._GetSize(ctx, self.GETSIZE_PATH, **kwargs)

    def testBasic(self) -> None:
        """Simple test."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", self.GETSIZE_PATH], stdout=StatTest.STAT_OUTPUT
        )
        self.assertEqual(self.GetSize(), 74)


class UnmockedGetSizeTest(cros_test_lib.TempDirTestCase):
    """Tests GetSize functionality w/out mocks."""

    @cros_test_lib.pytestmark_network_test
    def testBasic(self) -> None:
        """Simple test."""
        ctx = gs.GSContext()

        local_file = os.path.join(self.tempdir, "foo")
        osutils.WriteFile(local_file, "!" * 5)

        with gs.TemporaryURL("chromite.getsize") as tempuri:
            ctx.Copy(local_file, tempuri)
            self.assertEqual(ctx.GetSize(tempuri), 5)

    def testLocal(self) -> None:
        """Test local files."""
        ctx = gs.GSContext()
        f = os.path.join(self.tempdir, "f")

        osutils.Touch(f)
        self.assertEqual(ctx.GetSize(f), 0)

        osutils.WriteFile(f, "f" * 10)
        self.assertEqual(ctx.GetSize(f), 10)


class GetCreationTimeTest(AbstractGSContextTest):
    """Test GetCreationTime functionality."""

    def testBasic(self) -> None:
        """Simple test."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", GS_PACKAGES_PATH], stdout=STAT_OUTPUT_NOW
        )
        ctx = gs.GSContext()
        result = ctx.GetCreationTime(GS_PACKAGES_PATH)
        self.gs_mock.assertCommandContains(["stat", "--", GS_PACKAGES_PATH])
        self.assertEqual(result, NOW_DATETIME)

    def testURlNoExist(self) -> None:
        self.gs_mock.AddCmdResult(
            ["stat", "--", GS_PACKAGES_WRONG_PATH],
            stderr=STAT_OUTPUT_ERR,
            returncode=1,
        )
        ctx = gs.GSContext()
        self.assertRaises(
            gs.GSNoSuchKey, ctx.GetCreationTime, GS_PACKAGES_WRONG_PATH
        )
        self.gs_mock.assertCommandContains(
            ["stat", "--", GS_PACKAGES_WRONG_PATH]
        )


class UnMockedGetCreationTimeTest(cros_test_lib.TempDirTestCase):
    """Test GetCreationTime functionality without mocks."""

    @cros_test_lib.pytestmark_network_test
    def testGetCreationTime(self) -> None:
        """Test getting the creation time of a file."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("testGetCreationTime") as url:
            self.assertRaises(gs.GSNoSuchKey, ctx.GetCreationTime, url)

            ctx.CreateWithContents(url, "test file contents")

            result = ctx.GetCreationTime(url)
        self.assertIsInstance(result, datetime.datetime)


class GetCreationTimeSinceTest(AbstractGSContextTest):
    """Test GetCreationTimeSince functionality."""

    OLDER_RAW = NOW_RAW - datetime.timedelta(days=10)
    OLDER_STRING = datetime.datetime.strftime(
        OLDER_RAW.replace(tzinfo=datetime.timezone.utc), STAT_DATE_FORMAT
    )

    STAT_OUTPUT_OLDER = f"""{GS_PACKAGES_PATH}:
        Creation time:    {OLDER_STRING}
        Content-Language: en
        Content-Length:   74
        Content-Type:   application/octet-stream
        Hash (crc32c):    BBPMPA==
        Hash (md5):   ms+qSYvgI9SjXn8tW/5UpQ==
        ETag:     CNCgocbmqMACEAE=
        Generation:   1408776800850000
        Metageneration:   1
      """

    def testBasic(self) -> None:
        """Simple test."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", GS_PACKAGES_PATH], stdout=self.STAT_OUTPUT_OLDER
        )
        ctx = gs.GSContext()
        result = ctx.GetCreationTimeSince(GS_PACKAGES_PATH, NOW_RAW)
        self.gs_mock.assertCommandContains(["stat", "--", GS_PACKAGES_PATH])
        self.assertEqual(result.days, 10)

    def testURlNoExist(self) -> None:
        self.gs_mock.AddCmdResult(
            ["stat", "--", GS_PACKAGES_WRONG_PATH],
            stderr=STAT_OUTPUT_ERR,
            returncode=1,
        )
        ctx = gs.GSContext()
        with self.assertRaises(gs.GSNoSuchKey):
            ctx.GetCreationTimeSince(GS_PACKAGES_WRONG_PATH, self.OLDER_RAW)


class UnMockedGetCreationTimeSinceTest(cros_test_lib.TempDirTestCase):
    """Test GetCreationTimeSince functionality without mocks."""

    @cros_test_lib.pytestmark_network_test
    def testGetCreationTimeSince(self) -> None:
        """Test getting the creation time of a file."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("testGetCreationTime") as url:
            with self.assertRaises(gs.GSNoSuchKey):
                self.assertRaises(
                    gs.GSNoSuchKey, ctx.GetCreationTimeSince, url, NOW_RAW
                )
                ctx.GetCreationTimeSince(url, NOW_RAW)

            ctx.CreateWithContents(url, "test file contents")

            result = ctx.GetCreationTimeSince(url, NOW_RAW)
        self.assertIsInstance(result, datetime.timedelta)


class LSTest(AbstractGSContextTest):
    """Tests LS/List functionality."""

    LS_PATH = "gs://test/path/to/list"
    LS_OUTPUT_LINES = [
        "%s/foo" % LS_PATH,
        "%s/bar bell" % LS_PATH,
        "%s/nada/" % LS_PATH,
    ]
    LS_OUTPUT = "\n".join(LS_OUTPUT_LINES)

    SIZE1 = 12345
    SIZE2 = 654321
    DT1 = datetime.datetime(2000, 1, 2, 10, 10, 10)
    DT2 = datetime.datetime(2010, 3, 14)
    DT_STR1 = DT1.strftime(gs.DATETIME_FORMAT)
    DT_STR2 = DT2.strftime(gs.DATETIME_FORMAT)
    DETAILED_LS_OUTPUT_LINES = [
        "%10d  %s  %s/foo" % (SIZE1, DT_STR1, LS_PATH),
        "%10d  %s  %s/bar bell" % (SIZE2, DT_STR2, LS_PATH),
        "          %s/nada/" % LS_PATH,
        "TOTAL: 3 objects, XXXXX bytes (X.XX GB)",
    ]
    DETAILED_LS_OUTPUT = "\n".join(DETAILED_LS_OUTPUT_LINES)

    LIST_RESULT = [
        gs.GSListResult(
            content_length=SIZE1,
            creation_time=DT1,
            url="%s/foo" % LS_PATH,
            generation=None,
            metageneration=None,
        ),
        gs.GSListResult(
            content_length=SIZE2,
            creation_time=DT2,
            url="%s/bar bell" % LS_PATH,
            generation=None,
            metageneration=None,
        ),
        gs.GSListResult(
            content_length=None,
            creation_time=None,
            url="%s/nada/" % LS_PATH,
            generation=None,
            metageneration=None,
        ),
    ]

    def _LS(self, ctx, path, **kwargs):
        return ctx.LS(path, **kwargs)

    def LS(self, ctx=None, **kwargs):
        if ctx is None:
            ctx = self.ctx
        return self._LS(ctx, self.LS_PATH, **kwargs)

    def _List(self, ctx, path, **kwargs):
        return ctx.List(path, **kwargs)

    def List(self, ctx=None, **kwargs):
        if ctx is None:
            ctx = self.ctx
        return self._List(ctx, self.LS_PATH, **kwargs)

    def testBasicLS(self) -> None:
        """Simple LS test."""
        self.gs_mock.SetDefaultCmdResult(stdout=self.LS_OUTPUT)
        result = self.LS()
        self.gs_mock.assertCommandContains(["ls", "--", self.LS_PATH])

        self.assertEqual(self.LS_OUTPUT_LINES, result)

    def testBasicList(self) -> None:
        """Simple List test."""
        self.gs_mock.SetDefaultCmdResult(stdout=self.DETAILED_LS_OUTPUT)
        result = self.List(details=True)
        self.gs_mock.assertCommandContains(["ls", "-l", "--", self.LS_PATH])

        self.assertEqual(self.LIST_RESULT, result)


class UnmockedLSTest(cros_test_lib.TempDirTestCase):
    """Tests LS/List functionality w/out mocks."""

    def testLocalPaths(self) -> None:
        """Tests listing local paths."""
        ctx = gs.GSContext()

        # The tempdir should exist, but be empty, by default.
        self.assertEqual([], ctx.LS(self.tempdir))

        # Create a few random files.
        files = ["a", "b", "c!@", "d e f", "k\tj"]
        for f in files:
            osutils.Touch(os.path.join(self.tempdir, f))

        # See what the code finds -- order is not guaranteed.
        found = ctx.LS(self.tempdir)
        files.sort()
        found.sort()
        self.assertEqual(files, found)

    @cros_test_lib.pytestmark_network_test
    def testRemotePath(self) -> None:
        """Tests listing remote paths."""
        ctx = gs.GSContext()

        with gs.TemporaryURL("chromite.ls") as tempuri:
            # The path shouldn't exist by default.
            with self.assertRaises(gs.GSNoSuchKey):
                ctx.LS(tempuri)

            # Create some files with known sizes.
            files = ["a", "b", "c!@", "d e f", "k\tj"]
            uris = []
            for f in files:
                filename = os.path.join(self.tempdir, f)
                osutils.WriteFile(filename, f * 10)
                uri = os.path.join(tempuri, f)
                uris.append(uri)
                ctx.Copy(filename, uri)

            # Check the plain listing -- order is not guaranteed.
            found = ctx.LS(tempuri)
            uris.sort()
            found.sort()
            self.assertEqual(uris, found)

            # Check the detailed listing.
            found = ctx.List(tempuri, details=True)
            self.assertEqual(
                files, sorted([os.path.basename(x.url) for x in found])
            )

            # Check the detailed listing with multiple paths.
            found = ctx.List(
                ["%s/%s" % (tempuri, x) for x in files], details=True
            )
            self.assertEqual(
                files, sorted([os.path.basename(x.url) for x in found])
            )

            # Make sure sizes line up.
            for f in found:
                l = len(os.path.basename(f.url)) * 10
                self.assertEqual(f.content_length, l)
                self.assertIsNone(f.generation)
                self.assertIsNone(f.metageneration)

            # Check the generation listing with multiple paths.
            found = ctx.List([f"{tempuri}/{x}" for x in files], generation=True)
            self.assertGreater(len(found), 0)

            for f in found:
                self.assertGreater(f.generation, 0)
                self.assertGreater(f.metageneration, 0)


class CopyTest(AbstractGSContextTest, cros_test_lib.TempDirTestCase):
    """Tests GSContext.Copy() functionality."""

    GIVEN_REMOTE = EXPECTED_REMOTE = "gs://test/path/file"
    ACL = "public-read"

    def setUp(self) -> None:
        self.local_path = os.path.join(self.tempdir, "file")
        osutils.WriteFile(self.local_path, "")

    def _Copy(self, ctx, src, dst, **kwargs):
        return ctx.Copy(src, dst, **kwargs)

    def Copy(self, ctx=None, **kwargs):
        if ctx is None:
            ctx = self.ctx
        return self._Copy(ctx, self.local_path, self.GIVEN_REMOTE, **kwargs)

    def testBasic(self) -> None:
        """Simple copy test."""
        self.Copy()
        self.gs_mock.assertCommandContains(
            ["cp", "--", self.local_path, self.EXPECTED_REMOTE]
        )

    def testWithACL(self) -> None:
        """ACL specified during init."""
        ctx = gs.GSContext(acl=self.ACL)
        self.Copy(ctx=ctx)
        self.gs_mock.assertCommandContains(["cp", "-a", self.ACL])

    def testWithACL2(self) -> None:
        """ACL specified during invocation."""
        self.Copy(acl=self.ACL)
        self.gs_mock.assertCommandContains(["cp", "-a", self.ACL])

    def testWithACL3(self) -> None:
        """ACL specified during invocation that overrides init."""
        ctx = gs.GSContext(acl=self.ACL)
        self.Copy(ctx=ctx, acl=self.ACL)
        self.gs_mock.assertCommandContains(["cp", "-a", self.ACL])

    def testRunCommandError(self) -> None:
        """Test RunCommandError is propagated."""
        self.gs_mock.AddCmdResult(partial_mock.In("cp"), returncode=1)
        self.assertRaises(cros_build_lib.RunCommandError, self.Copy)

    def testGSContextPreconditionFailed(self) -> None:
        """GSContextPreconditionFailed is raised properly."""
        self.gs_mock.AddCmdResult(
            partial_mock.In("cp"),
            returncode=1,
            stderr=self.gs_mock.GSResponsePreconditionFailed,
        )
        self.assertRaises(gs.GSContextPreconditionFailed, self.Copy)

    def testNonRecursive(self) -> None:
        """Test non-recursive copy."""
        self.Copy(recursive=False)
        self.gs_mock.assertCommandContains(["-r"], expected=False)

    def testRecursive(self) -> None:
        """Test recursive copy."""
        self.Copy(recursive=True)
        self.gs_mock.assertCommandContains(["-r"], expected=False)
        self._Copy(self.ctx, self.tempdir, self.GIVEN_REMOTE, recursive=True)
        self.gs_mock.assertCommandContains(["cp", "-r"])

    def testCompress(self) -> None:
        """Test auto_compress behavior."""
        path = os.path.join(self.tempdir, "ok.txt")
        self._Copy(self.ctx, path, self.GIVEN_REMOTE, auto_compress=True)
        self.gs_mock.assertCommandContains(["-Z"], expected=True)

    def testGeneration(self) -> None:
        """Test generation return value."""
        exp_gen = 1413571271901000
        stderr = (
            "Copying file:///dev/null [Content-Type=application/octet-stream]"
            "...\n"
            "Uploading   %(uri)s:               0 B    \r"
            "Uploading   %(uri)s:               0 B    \r"
            "Created: %(uri)s#%(gen)s\n"
        ) % {"uri": self.GIVEN_REMOTE, "gen": exp_gen}
        self.gs_mock.AddCmdResult(
            partial_mock.In("cp"), returncode=0, stderr=stderr
        )
        gen = self.Copy()
        self.assertEqual(gen, exp_gen)

    def testGeneration404(self) -> None:
        """Test behavior when we get weird output."""
        stderr = (
            # This is a bit verbose, but it's from real output, so should be
            # fine.
            "Copying file:///tmp/tmpyUUPg1 [Content-Type=application/"
            "octet-stream]...\n"
            "Uploading   ...recovery-R38-6158.66.0-mccloud.instructions.lock:"
            " 0 B/38 B    \r"
            "Uploading   ...recovery-R38-6158.66.0-mccloud.instructions.lock:"
            " 38 B/38 B    \r"
            'NotFoundException: 404 Attempt to get key for "'
            "gs://chromeos-releases/tobesigned/50,beta-\n"
            "channel,mccloud,6158.66.0,ChromeOS-\n"
            'recovery-R38-6158.66.0-mccloud.instructions.lock" failed. '
            "This can happen if the\n"
            "URI refers to a non-existent object or if you meant to operate on "
            "a directory\n"
            "(e.g., leaving off -R option on gsutil cp, mv, or ls of a "
            "bucket)\n"
        )
        self.gs_mock.AddCmdResult(
            partial_mock.In("cp"), returncode=1, stderr=stderr
        )
        self.assertEqual(self.Copy(), None)


class UnmockedCopyTest(cros_test_lib.TempDirTestCase):
    """Tests Copy functionality w/out mocks."""

    @cros_test_lib.pytestmark_network_test
    def testNormal(self) -> None:
        """Test normal upload/download behavior."""
        ctx = gs.GSContext()

        content = "foooooooooooooooo!@!"

        local_src_file = os.path.join(self.tempdir, "src.txt")
        local_dst_file = os.path.join(self.tempdir, "dst.txt")

        osutils.WriteFile(local_src_file, content)

        with gs.TemporaryURL("chromite.cp") as tempuri:
            # Upload the file.
            gen = ctx.Copy(local_src_file, tempuri)

            # Verify the generation is valid.  All we can assume is that it's a
            # valid whole number greater than 0.
            self.assertNotEqual(gen, None)
            self.assertIsInstance(gen, numbers.Integral)
            self.assertGreater(gen, 0)

            # Verify the size is what we expect.
            self.assertEqual(
                ctx.GetSize(tempuri), os.path.getsize(local_src_file)
            )

            # Copy it back down and verify the content is unchanged.
            ctx.Copy(tempuri, local_dst_file)
            new_content = osutils.ReadFile(local_dst_file)
            self.assertEqual(content, new_content)

    @cros_test_lib.pytestmark_network_test
    def testCompress(self) -> None:
        """Test auto_compress behavior."""
        ctx = gs.GSContext()

        # Need a string that compresses well.
        content = (
            "zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz"
            "zzzzzlkasjdf89j2;3o4kqmnioruasddfv89uxdp;foiasjdf0892qn5kln"
        )

        local_src_file = os.path.join(self.tempdir, "src.txt")
        local_dst_file = os.path.join(self.tempdir, "dst.txt")

        osutils.WriteFile(local_src_file, content)

        with gs.TemporaryURL("chromite.cp") as tempuri:
            # Upload & compress the file.
            gen = ctx.Copy(local_src_file, tempuri, auto_compress=True)

            # Verify the generation is valid.  All we can assume is that it's a
            # valid whole number greater than 0.
            self.assertNotEqual(gen, None)
            self.assertGreater(gen, 0)

            # Verify the size is smaller (because it's compressed).
            self.assertLess(
                ctx.GetSize(tempuri), os.path.getsize(local_src_file)
            )

            # Copy it back down and verify the content is decompressed &
            # unchanged.
            ctx.Copy(tempuri, local_dst_file)
            new_content = osutils.ReadFile(local_dst_file)
            self.assertEqual(content, new_content)

    @cros_test_lib.pytestmark_network_test
    def testVersion(self) -> None:
        """Test version (generation) behavior."""
        ctx = gs.GSContext()

        local_src_file = os.path.join(self.tempdir, "src.txt")

        with gs.TemporaryURL("chromite.cp") as tempuri:
            # Upload the file.
            osutils.WriteFile(local_src_file, "gen0")
            gen = ctx.Copy(local_src_file, tempuri, version=0)

            # Verify the generation is valid.  All we can assume is that it's a
            # valid whole number greater than 0.
            self.assertNotEqual(gen, None)
            self.assertGreater(gen, 0)

            # The file should exist, so this will die due to wrong generation.
            osutils.WriteFile(local_src_file, "gen-bad")
            self.assertRaises(
                gs.GSContextPreconditionFailed,
                ctx.Copy,
                local_src_file,
                tempuri,
                version=0,
            )

            # Sanity check the content is unchanged.
            self.assertEqual(ctx.Cat(tempuri, encoding="utf-8"), "gen0")

            # Upload the file, but with the right generation.
            osutils.WriteFile(local_src_file, b"gen-new", mode="wb")
            gen = ctx.Copy(local_src_file, tempuri, version=gen)
            self.assertEqual(ctx.Cat(tempuri), b"gen-new")


class CopyIntoTest(CopyTest):
    """Test CopyInto functionality."""

    FILE = "ooga"
    GIVEN_REMOTE = "gs://test/path/file"
    EXPECTED_REMOTE = "%s/%s" % (GIVEN_REMOTE, FILE)

    def _Copy(self, ctx, src, dst, **kwargs):
        return ctx.CopyInto(src, dst, filename=self.FILE, **kwargs)


class RemoveTest(AbstractGSContextTest):
    """Tests GSContext.Remove() functionality."""

    def testNormal(self) -> None:
        """Test normal remove behavior."""
        self.assertEqual(self.ctx.Remove("gs://foo/bar"), None)

    def testMissing(self) -> None:
        """Test behavior w/missing files."""
        self.gs_mock.AddCmdResult(
            ["rm", "--", "gs://foo/bar"],
            stderr="CommandException: No URLs matched: " "gs://foo/bar",
            returncode=1,
        )
        self.assertRaises(gs.GSNoSuchKey, self.ctx.Remove, "gs://foo/bar")
        # This one should not throw an exception.
        self.ctx.Remove("gs://foo/bar", ignore_missing=True)

    def testRecursive(self) -> None:
        """Verify we pass down -R in recursive mode."""
        self.ctx.Remove("gs://foo/bar", recursive=True)
        self.gs_mock.assertCommandContains(["rm", "-R"])

    def testMultiple(self) -> None:
        """Test handling of multiple paths."""
        self.ctx.Remove(["gs://foo/bar", "gs://fat/cow"], recursive=True)
        self.gs_mock.assertCommandContains(
            ["rm", "-R", "--", "gs://foo/bar", "gs://fat/cow"]
        )


class UnmockedRemoveTest(cros_test_lib.TestCase):
    """Tests Remove functionality w/out mocks."""

    @cros_test_lib.pytestmark_network_test
    def testNormal(self) -> None:
        """Test normal remove behavior."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.rm") as tempuri:
            ctx.Copy("/dev/null", tempuri)
            self.assertEqual(ctx.Remove(tempuri), None)

    @cros_test_lib.pytestmark_network_test
    def testMissing(self) -> None:
        """Test behavior w/missing files."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.rm") as tempuri:
            self.assertRaises(gs.GSNoSuchKey, ctx.Remove, tempuri)
            # This one should not throw an exception.
            ctx.Remove(tempuri, ignore_missing=True)

    @cros_test_lib.pytestmark_network_test
    def testRecursive(self) -> None:
        """Verify recursive mode works."""
        files = ("a", "b/c", "d/e/ffff")
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.rm") as tempuri:
            for p in files:
                ctx.Copy("/dev/null", os.path.join(tempuri, p))
            ctx.Remove(tempuri, recursive=True)
            for p in files:
                self.assertFalse(ctx.Exists(os.path.join(tempuri, p)))

    @cros_test_lib.pytestmark_network_test
    def testMultiple(self) -> None:
        """Test handling of multiple paths."""
        files = ("a", "b/c", "d/e/ffff")
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.rm") as tempuri:
            for p in files:
                ctx.Copy("/dev/null", os.path.join(tempuri, p))
            ctx.Remove(["%s/%s" % (tempuri, x) for x in files])
            for p in files:
                self.assertFalse(ctx.Exists(os.path.join(tempuri, p)))

    @cros_test_lib.pytestmark_network_test
    def testGeneration(self) -> None:
        """Test conditional remove behavior."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.rm") as tempuri:
            ctx.Copy("/dev/null", tempuri)
            gen, _ = ctx.GetGeneration(tempuri)
            self.assertRaises(
                gs.GSContextPreconditionFailed,
                ctx.Remove,
                tempuri,
                version=gen + 1,
            )
            self.assertTrue(ctx.Exists(tempuri))
            ctx.Remove(tempuri, version=gen)
            self.assertFalse(ctx.Exists(tempuri))


class MoveTest(AbstractGSContextTest, cros_test_lib.TempDirTestCase):
    """Tests GSContext.Move() functionality."""

    GIVEN_REMOTE = EXPECTED_REMOTE = "gs://test/path/file"

    def setUp(self) -> None:
        self.local_path = os.path.join(self.tempdir, "file")
        osutils.WriteFile(self.local_path, "")

    def _Move(self, ctx, src, dst, **kwargs):
        return ctx.Move(src, dst, **kwargs)

    def Move(self, ctx=None, **kwargs):
        if ctx is None:
            ctx = self.ctx
        return self._Move(ctx, self.local_path, self.GIVEN_REMOTE, **kwargs)

    def testBasic(self) -> None:
        """Simple move test."""
        self.Move()
        self.gs_mock.assertCommandContains(
            ["mv", "--", self.local_path, self.EXPECTED_REMOTE]
        )


class GSContextInitTest(cros_test_lib.MockTempDirTestCase):
    """Tests GSContext.__init__() functionality."""

    def setUp(self) -> None:
        os.environ.pop("BOTO_CONFIG", None)
        self.bad_path = os.path.join(self.tempdir, "nonexistent")

        file_list = ["gsutil_bin", "boto_file", "acl_file"]
        cros_test_lib.CreateOnDiskHierarchy(self.tempdir, file_list)
        for f in file_list:
            setattr(self, f, os.path.join(self.tempdir, f))
        self.StartPatcher(PatchGS("DEFAULT_BOTO_FILE", new=self.boto_file))
        self.StartPatcher(PatchGS("_DEFAULT_GSUTIL_BIN", new=self.gsutil_bin))

    def testInitGsutilBin(self) -> None:
        """Test we use the given gsutil binary, erroring where appropriate."""
        # pylint: disable=protected-access
        gs.GSContext._CRCMOD_METHOD = "missing"
        self.assertEqual(
            gs.GSContext()._gsutil_bin, [sys.executable, self.gsutil_bin]
        )

        gs.GSContext._CRCMOD_METHOD = "vpython"
        self.assertEqual(
            gs.GSContext()._gsutil_bin, ["vpython3", self.gsutil_bin]
        )

        self.assertRaises(
            gs.GSContextException, gs.GSContext, gsutil_bin=self.bad_path
        )

        gs.GSContext._CRCMOD_METHOD = None

    def testBadGSUtilBin(self) -> None:
        """Test exception thrown for bad gsutil paths."""
        self.assertRaises(
            gs.GSContextException, gs.GSContext, gsutil_bin=self.bad_path
        )

    def testInitBotoFileEnv(self) -> None:
        """Test boto file environment is set correctly."""
        # We use gsutil_bin as a file that already exists and is not the
        # default.
        os.environ["BOTO_CONFIG"] = self.gsutil_bin
        self.assertTrue(gs.GSContext().boto_file, self.gsutil_bin)
        self.assertEqual(
            gs.GSContext(boto_file=self.acl_file).boto_file, self.acl_file
        )
        self.assertEqual(
            gs.GSContext(boto_file=self.bad_path).boto_file, self.bad_path
        )

    def testInitBotoFileEnvError(self) -> None:
        """Boto file through env var error."""
        self.assertEqual(gs.GSContext().boto_file, self.boto_file)
        # Check env usage next; no need to cleanup, teardown handles it,
        # and we want the env var to persist for the next part of this test.
        os.environ["BOTO_CONFIG"] = self.bad_path
        self.assertEqual(gs.GSContext().boto_file, self.bad_path)

    def testInitBotoFileError(self) -> None:
        """Test bad boto file."""
        self.assertEqual(
            gs.GSContext(boto_file=self.bad_path).boto_file, self.bad_path
        )

    def testDoNotUseDefaultBotoFileIfItDoesNotExist(self) -> None:
        """Do not set boto file if the default path does not exist."""
        if "BOTO_CONFIG" in os.environ:
            del os.environ["BOTO_CONFIG"]
        gs.GSContext.DEFAULT_BOTO_FILE = "foo/bar/doesnotexist"
        self.assertEqual(gs.GSContext().boto_file, None)

    def testInitAclFile(self) -> None:
        """Test ACL selection logic in __init__."""
        self.assertEqual(gs.GSContext().acl, None)
        self.assertEqual(gs.GSContext(acl=self.acl_file).acl, self.acl_file)

    def _testHTTPProxySettings(self, d) -> None:
        flags = gs.GSContext().gsutil_flags
        for key in d:
            flag = "Boto:%s=%s" % (key, d[key])
            error_msg = "%s not in %s" % (flag, " ".join(flags))
            self.assertTrue(flag in flags, error_msg)

    def testHTTPProxy(self) -> None:
        """Test we set http proxy correctly."""
        d = {
            "proxy": "fooserver",
            "proxy_user": "foouser",
            "proxy_pass": "foopasswd",
            "proxy_port": "8080",
        }
        os.environ["http_proxy"] = "http://%s:%s@%s:%s/" % (
            d["proxy_user"],
            d["proxy_pass"],
            d["proxy"],
            d["proxy_port"],
        )
        self._testHTTPProxySettings(d)

    def testHTTPProxyNoPort(self) -> None:
        """Test we accept http proxy without port number."""
        d = {
            "proxy": "fooserver",
            "proxy_user": "foouser",
            "proxy_pass": "foopasswd",
        }
        os.environ["http_proxy"] = "http://%s:%s@%s/" % (
            d["proxy_user"],
            d["proxy_pass"],
            d["proxy"],
        )
        self._testHTTPProxySettings(d)

    def testHTTPProxyNoUserPasswd(self) -> None:
        """Test we accept http proxy without user and password."""
        d = {"proxy": "fooserver", "proxy_port": "8080"}
        os.environ["http_proxy"] = "http://%s:%s/" % (
            d["proxy"],
            d["proxy_port"],
        )
        self._testHTTPProxySettings(d)

    def testHTTPProxyNoPasswd(self) -> None:
        """Test we accept http proxy without password."""
        d = {
            "proxy": "fooserver",
            "proxy_user": "foouser",
            "proxy_port": "8080",
        }
        os.environ["http_proxy"] = "http://%s@%s:%s/" % (
            d["proxy_user"],
            d["proxy"],
            d["proxy_port"],
        )
        self._testHTTPProxySettings(d)


class GSDoCommandTest(cros_test_lib.TestCase):
    """Tests of gs.DoCommand behavior.

    This test class inherits from cros_test_lib.TestCase instead of from
    AbstractGSContextTest, because the latter unnecessarily mocks out
    cros_build_lib.run, in a way that breaks _testDoCommand (changing
    cros_build_lib.run to refer to a mock instance after the
    GenericRetry mock has already been set up to expect a reference to the
    original run).
    """

    def setUp(self) -> None:
        self.ctx = gs.GSContext()

    def _testDoCommand(
        self,
        ctx,
        headers=(),
        retries=None,
        sleep=None,
        version=None,
        recursive=False,
    ) -> None:
        if retries is None:
            retries = ctx.DEFAULT_RETRIES
        if sleep is None:
            sleep = ctx.DEFAULT_SLEEP_TIME

        result = cros_build_lib.CompletedProcess(stderr="")
        with mock.patch.object(
            retry_stats, "RetryWithStats", autospec=True, return_value=result
        ):
            ctx.Copy("/blah", "gs://foon", version=version, recursive=recursive)
            # pylint: disable=protected-access
            cmd = self.ctx._gsutil_bin + self.ctx.gsutil_flags + list(headers)
            cmd += ["cp", "-v"]
            if recursive:
                cmd += ["-r", "-e"]
            cmd += ["--", "/blah", "gs://foon"]

            # pylint: disable=protected-access
            retry_stats.RetryWithStats.assert_called_once_with(
                retry_stats.GSUTIL,
                ctx._RetryFilter,
                retries,
                cros_build_lib.run,
                cmd,
                sleep=sleep,
                stderr=True,
                stdout=True,
                encoding="utf-8",
                extra_env=mock.ANY,
            )

    def testDoCommandDefault(self) -> None:
        """Verify the internal DoCommand function works correctly."""
        self._testDoCommand(self.ctx)

    def testDoCommandCustom(self) -> None:
        """Test that retries and sleep parameters are honored."""
        ctx = gs.GSContext(retries=4, sleep=1)
        self._testDoCommand(ctx, retries=4, sleep=1)

    def testVersion(self) -> None:
        """Test that the version field expands into the header."""
        self._testDoCommand(
            self.ctx, version=3, headers=["-h", "x-goog-if-generation-match:3"]
        )

    def testDoCommandRecursiveCopy(self) -> None:
        """Test that recursive copy command is honored."""
        self._testDoCommand(self.ctx, recursive=True)


class GSRetryFilterTest(cros_test_lib.TestCase):
    """Verifies that we filter and process gsutil errors correctly."""

    # pylint: disable=protected-access

    LOCAL_PATH = "/tmp/file"
    REMOTE_PATH = (
        "gs://chromeos-prebuilt/board/beltino/paladin-R33-4926.0.0"
        "-rc2/packages/chromeos-base/autotest-tests-0.0.1-r4679.tbz2"
    )
    GSUTIL_TRACKER_DIR = "/foo"
    UPLOAD_TRACKER_FILE = (
        "upload_TRACKER_9263880a80e4a582aec54eaa697bfcdd9c5621ea.9.tbz2__JSON."
        "url"
    )
    DOWNLOAD_TRACKER_FILE = (
        "download_TRACKER_5a695131f3ef6e4c903f594783412bb996a7f375._file__JSON."
        "etag"
    )
    RETURN_CODE = 3

    def setUp(self) -> None:
        self.ctx = gs.GSContext()
        self.ctx.DEFAULT_GSUTIL_TRACKER_DIR = self.GSUTIL_TRACKER_DIR

    def _getException(self, cmd, error, returncode=RETURN_CODE):
        result = cros_build_lib.CompletedProcess(
            args=cmd, stderr=error, returncode=returncode
        )
        return cros_build_lib.RunCommandError("blah", result)

    def assertNoSuchKey(self, error_msg) -> None:
        cmd = ["gsutil", "ls", self.REMOTE_PATH]
        e = self._getException(cmd, error_msg)
        self.assertRaises(gs.GSNoSuchKey, self.ctx._RetryFilter, e)

    def assertPreconditionFailed(self, error_msg) -> None:
        cmd = ["gsutil", "ls", self.REMOTE_PATH]
        e = self._getException(cmd, error_msg)
        self.assertRaises(
            gs.GSContextPreconditionFailed, self.ctx._RetryFilter, e
        )

    def testRetryOnlyFlakyErrors(self) -> None:
        """Test that we retry only flaky errors."""
        cmd = ["gsutil", "ls", self.REMOTE_PATH]
        e = self._getException(cmd, "ServiceException: 503")
        self.assertTrue(self.ctx._RetryFilter(e))

        e = self._getException(cmd, "UnknownException: 603")
        self.assertFalse(self.ctx._RetryFilter(e))

    def testRaiseGSErrors(self) -> None:
        """Test that we raise appropriate exceptions."""
        self.assertNoSuchKey("CommandException: No URLs matched.")
        self.assertNoSuchKey("NotFoundException: 404")
        self.assertPreconditionFailed(
            "PreconditionException: 412 Precondition Failed"
        )

    @mock.patch("chromite.lib.osutils.SafeUnlink")
    @mock.patch("chromite.lib.osutils.ReadFile")
    @mock.patch("os.path.exists")
    def testRemoveUploadTrackerFile(
        self, exists_mock, readfile_mock, unlink_mock
    ) -> None:
        """Test removal of tracker files for resumable upload failures."""
        cmd = ["gsutil", "cp", self.LOCAL_PATH, self.REMOTE_PATH]
        e = self._getException(cmd, self.ctx.RESUMABLE_UPLOAD_ERROR)
        exists_mock.return_value = True
        readfile_mock.return_value = "foohash"
        self.ctx._RetryFilter(e)
        tracker_file_path = os.path.join(
            self.GSUTIL_TRACKER_DIR, self.UPLOAD_TRACKER_FILE
        )
        unlink_mock.assert_called_once_with(tracker_file_path)

    @mock.patch("chromite.lib.osutils.SafeUnlink")
    @mock.patch("chromite.lib.osutils.ReadFile")
    @mock.patch("os.path.exists")
    def testRemoveDownloadTrackerFile(
        self, exists_mock, readfile_mock, unlink_mock
    ) -> None:
        """Test removal of tracker files for resumable download failures."""
        cmd = ["gsutil", "cp", self.REMOTE_PATH, self.LOCAL_PATH]
        e = self._getException(cmd, self.ctx.RESUMABLE_DOWNLOAD_ERROR)
        exists_mock.return_value = True
        readfile_mock.return_value = "foohash"
        self.ctx._RetryFilter(e)
        tracker_file_path = os.path.join(
            self.GSUTIL_TRACKER_DIR, self.DOWNLOAD_TRACKER_FILE
        )
        unlink_mock.assert_called_once_with(tracker_file_path)

    def testRemoveTrackerFileOnlyForCP(self) -> None:
        """Test that we remove tracker files only for 'gsutil cp'."""
        cmd = ["gsutil", "ls", self.REMOTE_PATH]
        e = self._getException(cmd, self.ctx.RESUMABLE_DOWNLOAD_ERROR)

        with mock.patch.object(gs.GSContext, "GetTrackerFilenames"):
            self.ctx._RetryFilter(e)
            self.assertFalse(self.ctx.GetTrackerFilenames.called)

    def testNoRemoveTrackerFileOnOtherErrors(self) -> None:
        """Verify we do not attempt to delete tracker files for other errors."""
        cmd = ["gsutil", "cp", self.REMOTE_PATH, self.LOCAL_PATH]
        e = self._getException(cmd, "One or more URLs matched no objects")

        with mock.patch.object(gs.GSContext, "GetTrackerFilenames"):
            self.assertRaises(gs.GSNoSuchKey, self.ctx._RetryFilter, e)
            self.assertFalse(self.ctx.GetTrackerFilenames.called)

    def testRetryTransient(self) -> None:
        """Verify retry behavior when hitting b/11762375"""
        error = (
            "Removing gs://foo/bar/monkey...\n"
            "GSResponseError: status=403, code=InvalidAccessKeyId, "
            'reason="Forbidden", message="The User Id you provided '
            'does not exist in our records.", detail="GOOGBWPADTH7OV25KJXZ"'
        )
        e = self._getException(["gsutil", "rm", "gs://foo/bar/monkey"], error)
        self.assertEqual(self.ctx._RetryFilter(e), True)

    def testRetrySSLEOF(self) -> None:
        """Verify retry behavior on EOF in violation of SSL protocol."""
        error = (
            "ssl.SSLError: [Errno 8] _ssl.c:510: EOF occurred in violation of"
            " protocol"
        )
        e = self._getException(
            ["gsutil", "cat", "gs://totally/legit/uri"], error
        )
        self.assertEqual(self.ctx._RetryFilter(e), True)

    def testRetrySSLTimeout(self) -> None:
        """Verify retry behavior when read operation timed out."""
        error = "ssl.SSLError: ('The read operation timed out',)"
        e = self._getException(
            ["gsutil", "cp", self.REMOTE_PATH, self.LOCAL_PATH], error
        )
        self.assertEqual(self.ctx._RetryFilter(e), True)

    def testRetrySSLHandshakeTimeout(self) -> None:
        """Verify retry behavior when handshake operation timed out."""
        error = "ssl.SSLError: _ssl.c:495: The handshake operation timed out"
        e = self._getException(
            ["gsutil", "cp", self.REMOTE_PATH, self.LOCAL_PATH], error
        )
        self.assertEqual(self.ctx._RetryFilter(e), True)

    def testRetryAccessDeniedException(self) -> None:
        """Verify retry behavior on transient AccessDeniedException."""
        error = (
            "AccessDeniedException: 403 XXX@gmail.com does not have "
            "storage.objects.delete access to XXX"
            "CommandException: 1 file/object could not be transferred."
        )
        e = self._getException(
            ["gsutil", "cp", self.REMOTE_PATH, self.LOCAL_PATH], error
        )
        self.assertEqual(self.ctx._RetryFilter(e), True)


class GSContextTest(AbstractGSContextTest):
    """Tests for GSContext()"""

    URL = "gs://chromeos-image-archive/x86-mario-release/R17-1413.0.0-a1-b1346"
    FILE_NAME = "chromeos_R17-1413.0.0-a1_x86-mario_full_dev.bin"
    GS_PATH = "gs://test/path/to/list"
    DT = datetime.datetime(2000, 1, 2, 10, 10, 10)
    DT_STR = DT.strftime(gs.DATETIME_FORMAT)
    DETAILED_LS_OUTPUT_LINES = [
        "%10d  %s  %s/mock_data" % (100, DT_STR, GS_PATH),
        "%10d  %s  %s/mock_data" % (100, DT_STR, GS_PATH),
        "%10d  %s  %s/%s" % (100, DT_STR, GS_PATH, FILE_NAME),
        "TOTAL: 3 objects, XXXXX bytes (X.XX GB)",
    ]

    LIST_RESULT = [
        gs.GSListResult(
            content_length=100,
            creation_time=DT,
            url="%s/mock_data" % GS_PATH,
            generation=None,
            metageneration=None,
        ),
        gs.GSListResult(
            content_length=100,
            creation_time=DT,
            url="%s/mock_data" % GS_PATH,
            generation=None,
            metageneration=None,
        ),
        gs.GSListResult(
            content_length=100,
            creation_time=DT,
            url="%s/%s" % (GS_PATH, FILE_NAME),
            generation=None,
            metageneration=None,
        ),
    ]

    def testTemporaryUrl(self) -> None:
        """Just verify the url helper generates valid URLs."""
        with gs.TemporaryURL("mock") as url:
            base = url[0 : len(constants.TRASH_BUCKET)]
            self.assertEqual(base, constants.TRASH_BUCKET)

            valid_chars = set(string.ascii_letters + string.digits + "/-")
            used_chars = set(url[len(base) + 1 :])
            self.assertEqual(used_chars - valid_chars, set())

    def testSetAclError(self) -> None:
        """Ensure SetACL blows up if the acl isn't specified."""
        self.assertRaises(gs.GSContextException, self.ctx.SetACL, "gs://abc/3")

    def testSetDefaultAcl(self) -> None:
        """Test default ACL behavior."""
        self.ctx.SetACL("gs://abc/1", "monkeys")
        self.gs_mock.assertCommandContains(
            ["acl", "set", "--", "monkeys", "gs://abc/1"]
        )

    def testSetAcl(self) -> None:
        """Base ACL setting functionality."""
        ctx = gs.GSContext(acl="/my/file/acl")
        ctx.SetACL("gs://abc/1")
        self.gs_mock.assertCommandContains(
            ["acl", "set", "/my/file/acl", "gs://abc/1"]
        )

    def testSetAclMultiple(self) -> None:
        """Test multiple paths at once."""
        ctx = gs.GSContext(acl="/my/file/acl")
        ctx.SetACL(["gs://abc/1", "gs://abc/2"])
        self.gs_mock.assertCommandContains(
            ["acl", "set", "--", "/my/file/acl", "gs://abc/1", "gs://abc/2"]
        )

    def testChangeAcl(self) -> None:
        """Test changing an ACL."""
        basic_file = """
-g foo:READ

-u bar:FULL_CONTROL"""
        comment_file = """
# Give foo READ permission
-g foo:READ # Now foo can read this
  # This whole line should be removed
-u bar:FULL_CONTROL
# A comment at the end"""
        tempfile = os.path.join(self.tempdir, "tempfile")
        ctx = gs.GSContext()

        osutils.WriteFile(tempfile, basic_file)
        ctx.ChangeACL("gs://abc/1", acl_args_file=tempfile)
        self.gs_mock.assertCommandContains(
            [
                "acl",
                "ch",
                "-g",
                "foo:READ",
                "-u",
                "bar:FULL_CONTROL",
                "gs://abc/1",
            ]
        )

        osutils.WriteFile(tempfile, comment_file)
        ctx.ChangeACL("gs://abc/1", acl_args_file=tempfile)
        self.gs_mock.assertCommandContains(
            [
                "acl",
                "ch",
                "-g",
                "foo:READ",
                "-u",
                "bar:FULL_CONTROL",
                "gs://abc/1",
            ]
        )

        ctx.ChangeACL(
            "gs://abc/1", acl_args=["-g", "foo:READ", "-u", "bar:FULL_CONTROL"]
        )
        self.gs_mock.assertCommandContains(
            [
                "acl",
                "ch",
                "-g",
                "foo:READ",
                "-u",
                "bar:FULL_CONTROL",
                "gs://abc/1",
            ]
        )

        with self.assertRaises(gs.GSContextException):
            ctx.ChangeACL(
                "gs://abc/1", acl_args_file=tempfile, acl_args=["foo"]
            )

        with self.assertRaises(gs.GSContextException):
            ctx.ChangeACL("gs://abc/1")

    def testIncrement(self) -> None:
        """Test ability to atomically increment a counter."""
        ctx = gs.GSContext()

        with mock.patch.object(ctx, "GetGeneration", return_value=(0, 0)):
            ctx.Counter("gs://abc/1").Increment()

        self.gs_mock.assertCommandContains(["cp", "gs://abc/1"])

    def testGetGeneration(self) -> None:
        """Test ability to get the generation of a file."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", "gs://abc/1"], stdout=StatTest.STAT_OUTPUT
        )
        ctx = gs.GSContext()
        ctx.GetGeneration("gs://abc/1")
        self.gs_mock.assertCommandContains(["stat", "--", "gs://abc/1"])

    def testCreateCached(self) -> None:
        """Test that the function runs through."""
        gs.GSContext(cache_dir=self.tempdir)

    def testReuseCached(self) -> None:
        """Test that second fetch is a cache hit."""
        gs.GSContext(cache_dir=self.tempdir)
        gs.GSUTIL_URL = None
        gs.GSContext(cache_dir=self.tempdir)

    def testUnknownError(self) -> None:
        """Verify when gsutil fails in an unknown way, we do the right thing."""
        self.gs_mock.AddCmdResult(["cat", "/asdf"], returncode=1)

        ctx = gs.GSContext()
        self.assertRaises(gs.GSCommandError, ctx.DoCommand, ["cat", "/asdf"])

    def testWaitForGsPathsAllPresent(self) -> None:
        """Test for waiting when all paths exist already."""
        ctx = gs.GSContext()

        with mock.patch.object(ctx, "Exists", return_value=True):
            ctx.WaitForGsPaths(["/path1", "/path2"], 20)

    def testWaitForGsPathsDelayedSuccess(self) -> None:
        """Test for waiting, but not all paths exist so we timeout."""
        ctx = gs.GSContext()

        # First they both don't exist, then one does, then remaining does.
        exists = [False, False, True, False, True]
        with mock.patch.object(ctx, "Exists", side_effect=exists):
            ctx.WaitForGsPaths(["/path1", "/path2"], 20, period=0.02)

    def testWaitForGsPathsTimeout(self) -> None:
        """Test for waiting, but not all paths exist so we timeout."""
        ctx = gs.GSContext()

        exists = {"/path1": True, "/path2": False}
        with mock.patch.object(ctx, "Exists", side_effect=lambda p: exists[p]):
            self.assertRaises(
                gs.timeout_util.TimeoutError,
                ctx.WaitForGsPaths,
                ["/path1", "/path2"],
                timeout=1,
                period=0.02,
            )

    def testParallelFalse(self) -> None:
        """Tests that "-m" is not used by default."""
        ctx = gs.GSContext()
        ctx.Copy("-", "gs://abc/1")
        self.assertFalse(any("-m" in cmd for cmd in self.gs_mock.raw_gs_cmds))

    def testParallelTrue(self) -> None:
        """Tests that "-m" is used when you pass parallel=True."""
        ctx = gs.GSContext()
        ctx.Copy("gs://abc/1", "gs://abc/2", parallel=True)
        self.assertTrue(all("-m" in cmd for cmd in self.gs_mock.raw_gs_cmds))

    def testNoParallelOpWithStdin(self) -> None:
        """Tests that "-m" is not used when we pipe the input."""
        ctx = gs.GSContext()
        ctx.Copy("gs://abc/1", "gs://abc/2", input="foo", parallel=True)
        self.assertFalse(any("-m" in cmd for cmd in self.gs_mock.raw_gs_cmds))

    def testGetGsNamesWithWait(self) -> None:
        """Test that we get the target artifact that is available."""
        pattern = "*_full_*"

        ctx = gs.GSContext()

        # GSUtil ls gs://archive_url_prefix/.
        self.gs_mock.SetDefaultCmdResult(
            stdout="\n".join(self.DETAILED_LS_OUTPUT_LINES)
        )

        # Timeout explicitly set to 0 to test that we always run at least once.
        result = ctx.GetGsNamesWithWait(pattern, self.URL, period=1, timeout=0)
        self.assertEqual([self.FILE_NAME], result)

    def testGetGsNamesWithWaitWithDirectStat(self) -> None:
        """Verify direct stat an artifact whose name is fully spelled out."""
        pattern = self.FILE_NAME

        ctx = gs.GSContext()

        exists = {"%s/%s" % (self.URL, self.FILE_NAME): True}
        with mock.patch.object(ctx, "Exists", side_effect=lambda p: exists[p]):
            # Timeout explicitly set to 0 to test that we always run at least
            # once.
            result = ctx.GetGsNamesWithWait(
                pattern, self.URL, period=1, timeout=0
            )
            self.assertEqual([self.FILE_NAME], result)

    def testGetGsNamesWithWaitWithRetry(self) -> None:
        """Test that we can poll until all target artifacts are available."""
        pattern = "*_full_*"

        ctx = gs.GSContext()

        # GSUtil ls gs://archive_url_prefix/.
        exists = [[], self.LIST_RESULT]
        with mock.patch.object(ctx, "List", side_effect=exists):
            # Timeout explicitly set to 0 to test that we always run at least
            # once.
            result = ctx.GetGsNamesWithWait(
                pattern, self.URL, period=1, timeout=4
            )
            self.assertEqual([self.FILE_NAME], result)

    def testGetGsNamesWithWaitTimeout(self) -> None:
        """Test that we can poll until all target artifacts are available."""
        pattern = "*_full_*"

        ctx = gs.GSContext()

        # GSUtil ls gs://archive_url_prefix/.
        self.gs_mock.SetDefaultCmdResult(stdout=[])

        # Timeout explicitly set to 0 to test that we always run at least once.
        result = ctx.GetGsNamesWithWait(pattern, self.URL, period=2, timeout=1)
        self.assertEqual(None, result)


class UnmockedGSContextTest(cros_test_lib.TempDirTestCase):
    """Tests for GSContext that go over the network."""

    @cros_test_lib.pytestmark_network_test
    def testIncrement(self) -> None:
        ctx = gs.GSContext()
        with gs.TemporaryURL("testIncrement") as url:
            counter = ctx.Counter(url)
            self.assertEqual(0, counter.Get())
            for i in range(1, 4):
                self.assertEqual(i, counter.Increment())
                self.assertEqual(i, counter.Get())

    @cros_test_lib.pytestmark_network_test
    def testGetGsNamesWithWait(self) -> None:
        """Tests getting files from remote paths."""
        file_name = "chromeos_R17-1413.0.0-a1_x86-mario_full_dev.bin"
        pattern = "*_full_*"

        ctx = gs.GSContext()

        with gs.TemporaryURL("testGetGsNamesWithWait") as url:
            # The path shouldn't exist by default.
            with self.assertRaises(gs.GSNoSuchKey):
                ctx.GetGsNamesWithWait(pattern, url, period=2, timeout=1)

            # Create files in bucket.
            files = ["mock_data", file_name]
            for f in files:
                filename = os.path.join(self.tempdir, f)
                osutils.WriteFile(filename, f * 10)
                uri = os.path.join(url, f)
                ctx.Copy(filename, uri)

            # Use pattern to get google storage names.
            self.assertEqual(
                ctx.GetGsNamesWithWait(pattern, url, period=1, timeout=0),
                [file_name],
            )
            # Use direct file name to get google storage names.
            self.assertEqual(
                ctx.GetGsNamesWithWait(file_name, url, period=1, timeout=0),
                [file_name],
            )

            # Remove the matched file, verify that GetGsNamesWithWait returns
            # None.
            ctx.Remove(os.path.join(url, file_name), ignore_missing=True)

            self.assertEqual(
                ctx.GetGsNamesWithWait(pattern, url, period=1, timeout=3), None
            )
            self.assertEqual(
                ctx.GetGsNamesWithWait(file_name, url, period=1, timeout=3),
                None,
            )


class StatTest(AbstractGSContextTest):
    """Tests Stat functionality."""

    # Convenient constant for mocking Stat results.
    STAT_OUTPUT = b"""gs://abc/1:
        Creation time:    Sat, 23 Aug 2014 06:53:20 GMT
        Content-Language: en
        Content-Length:   74
        Content-Type:   application/octet-stream
        Hash (crc32c):    BBPMPA==
        Hash (md5):   ms+qSYvgI9SjXn8tW/5UpQ==
        ETag:     CNCgocbmqMACEAE=
        Generation:   1408776800850000
        Metageneration:   1
      """

    # Stat output can vary based on how/when the file was created.
    STAT_OUTPUT_OLDER = b"""gs://abc/1:
        Creation time:    Sat, 23 Aug 2014 06:53:20 GMT
        Content-Length:   74
        Content-Type:   application/octet-stream
        Hash (crc32c):    BBPMPA==
        Hash (md5):   ms+qSYvgI9SjXn8tW/5UpQ==
        ETag:     CNCgocbmqMACEAE=
        Generation:   1408776800850000
        Metageneration:   1
      """

    # Stat output with no MD5 (this is not guaranteed by GS, and
    # can be omitted if files are uploaded as composite objects).
    STAT_OUTPUT_NO_MD5 = b"""gs://abc/1:
        Creation time:    Sat, 23 Aug 2014 06:53:20 GMT
        Content-Language: en
        Content-Length:   74
        Content-Type:   application/octet-stream
        Hash (crc32c):    BBPMPA==
        ETag:     CNCgocbmqMACEAE=
        Generation:   1408776800850000
        Metageneration:   1
      """

    # When stat throws an error.  It's a special snow flake.
    STAT_ERROR_OUTPUT = b"No URLs matched gs://abc/1"
    RETRY_STAT_ERROR_OUTPUT = (
        b"Retrying request, attempt #1...\nNo URLs matched gs://abc/1"
    )

    def testStat(self) -> None:
        """Test ability to get the generation of a file."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", "gs://abc/1"], stdout=self.STAT_OUTPUT
        )
        ctx = gs.GSContext()
        result = ctx.Stat("gs://abc/1")
        self.gs_mock.assertCommandContains(["stat", "--", "gs://abc/1"])

        self.assertEqual(
            result.creation_time, datetime.datetime(2014, 8, 23, 6, 53, 20)
        )
        self.assertEqual(result.content_length, 74)
        self.assertEqual(result.content_type, "application/octet-stream")
        self.assertEqual(result.hash_crc32c, "BBPMPA==")
        self.assertEqual(result.hash_md5, "ms+qSYvgI9SjXn8tW/5UpQ==")
        self.assertEqual(result.etag, "CNCgocbmqMACEAE=")
        self.assertEqual(result.generation, 1408776800850000)
        self.assertEqual(result.metageneration, 1)

    def testStatOlderOutput(self) -> None:
        """Test ability to get the generation of a file."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", "gs://abc/1"], stdout=self.STAT_OUTPUT_OLDER
        )
        ctx = gs.GSContext()
        result = ctx.Stat("gs://abc/1")
        self.gs_mock.assertCommandContains(["stat", "--", "gs://abc/1"])

        self.assertEqual(
            result.creation_time, datetime.datetime(2014, 8, 23, 6, 53, 20)
        )
        self.assertEqual(result.content_length, 74)
        self.assertEqual(result.content_type, "application/octet-stream")
        self.assertEqual(result.hash_crc32c, "BBPMPA==")
        self.assertEqual(result.hash_md5, "ms+qSYvgI9SjXn8tW/5UpQ==")
        self.assertEqual(result.etag, "CNCgocbmqMACEAE=")
        self.assertEqual(result.generation, 1408776800850000)
        self.assertEqual(result.metageneration, 1)

    def testStatNoMD5(self) -> None:
        """Make sure GSContext works without an MD5."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", "gs://abc/1"], stdout=self.STAT_OUTPUT_NO_MD5
        )
        ctx = gs.GSContext()
        result = ctx.Stat("gs://abc/1")
        self.gs_mock.assertCommandContains(["stat", "--", "gs://abc/1"])

        self.assertEqual(
            result.creation_time, datetime.datetime(2014, 8, 23, 6, 53, 20)
        )
        self.assertEqual(result.content_length, 74)
        self.assertEqual(result.content_type, "application/octet-stream")
        self.assertEqual(result.hash_crc32c, "BBPMPA==")
        self.assertEqual(result.hash_md5, None)
        self.assertEqual(result.etag, "CNCgocbmqMACEAE=")
        self.assertEqual(result.generation, 1408776800850000)
        self.assertEqual(result.metageneration, 1)

    def testStatNoExist(self) -> None:
        """Test ability to get the generation of a file."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", "gs://abc/1"],
            stderr=self.STAT_ERROR_OUTPUT,
            returncode=1,
        )
        ctx = gs.GSContext()
        self.assertRaises(gs.GSNoSuchKey, ctx.Stat, "gs://abc/1")
        self.gs_mock.assertCommandContains(["stat", "--", "gs://abc/1"])

    def testStatRetryNoExist(self) -> None:
        """Test ability to get the generation of a file."""
        self.gs_mock.AddCmdResult(
            ["stat", "--", "gs://abc/1"],
            stderr=self.RETRY_STAT_ERROR_OUTPUT,
            returncode=1,
        )
        ctx = gs.GSContext()
        self.assertRaises(gs.GSNoSuchKey, ctx.Stat, "gs://abc/1")
        self.gs_mock.assertCommandContains(["stat", "--", "gs://abc/1"])


class UnmockedStatTest(cros_test_lib.TempDirTestCase):
    """Tests Stat functionality w/out mocks."""

    @cros_test_lib.pytestmark_network_test
    def testStat(self) -> None:
        """Test ability to get the generation of a file."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("testStat") as url:
            # The URL doesn't exist. Test Stat for this case.
            self.assertRaises(gs.GSNoSuchKey, ctx.Stat, url)

            # Populate the URL.
            ctx.CreateWithContents(url, "test file contents")

            # Stat a URL that exists.
            result = ctx.Stat(url)

        # Verify the Stat results.
        self.assertIsInstance(result.creation_time, datetime.datetime)
        self.assertEqual(result.content_length, 18)
        self.assertEqual(result.content_type, "application/octet-stream")
        self.assertEqual(result.hash_crc32c, "wUc4sQ==")
        self.assertEqual(result.hash_md5, "iRvNNwBhmvUVG/lbg2/5sQ==")
        self.assertIsInstance(result.etag, str)
        self.assertIsInstance(result.generation, int)
        self.assertEqual(result.metageneration, 1)

    @cros_test_lib.pytestmark_network_test
    def testMissing(self) -> None:
        """Test exceptions when the file doesn't exist."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("testStat") as url:
            self.assertRaises(gs.GSNoSuchKey, ctx.Stat, url)
            self.assertFalse(ctx.Exists(url))

    def testExists(self) -> None:
        """Test Exists behavior with local files."""
        ctx = gs.GSContext()
        f = os.path.join(self.tempdir, "f")

        self.assertFalse(ctx.Exists(f))

        osutils.Touch(f)
        self.assertTrue(ctx.Exists(f))


class CatTest(cros_test_lib.TempDirTestCase):
    """Tests GSContext.Copy() functionality."""

    def testLocalFile(self) -> None:
        """Tests catting a local file."""
        ctx = gs.GSContext()
        filename = os.path.join(self.tempdir, "myfile")
        content = "foo"
        osutils.WriteFile(filename, content)
        self.assertEqual(content, ctx.Cat(filename, encoding="utf-8"))
        self.assertEqual(content.encode("utf-8"), ctx.Cat(filename))

    def testLocalMissingFile(self) -> None:
        """Tests catting a missing local file."""
        ctx = gs.GSContext()
        with self.assertRaises(gs.GSNoSuchKey):
            ctx.Cat(os.path.join(self.tempdir, "does/not/exist"))

    def testLocalForbiddenFile(self) -> None:
        """Tests catting a local file that we don't have access to."""
        ctx = gs.GSContext()
        filename = os.path.join(self.tempdir, "myfile")
        content = "foo"
        osutils.WriteFile(filename, content)
        os.chmod(filename, 0o000)
        with self.assertRaises(gs.GSContextException):
            ctx.Cat(filename)

    @cros_test_lib.pytestmark_network_test
    def testNetworkFile(self) -> None:
        """Tests catting a GS file."""
        ctx = gs.GSContext()
        filename = os.path.join(self.tempdir, "myfile")
        content = "fOoOoOoo1\n\thi@!*!(\r\r\nend"
        osutils.WriteFile(filename, content)

        with gs.TemporaryURL("chromite.cat") as tempuri:
            ctx.Copy(filename, tempuri)
            self.assertEqual(content, ctx.Cat(tempuri, encoding="utf-8"))

    @cros_test_lib.pytestmark_network_test
    def testNetworkMissingFile(self) -> None:
        """Tests catting a missing GS file."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.cat") as tempuri:
            with self.assertRaises(gs.GSNoSuchKey):
                ctx.Cat(tempuri)

    @cros_test_lib.pytestmark_network_test
    def testStreamingRemoteFile(self) -> None:
        """Test streaming a remote file."""
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.cat") as url:
            # The default chunksize is 0x100000 (1MB).
            first_chunk = b"a" * 0x100000
            second_chunk = b"aaaaaaaaabbbbbbccc"
            ctx.CreateWithContents(url, first_chunk + second_chunk)

            result = ctx.StreamingCat(url)
            # Get the 1st chunk.
            self.assertEqual(next(result), first_chunk)
            # Then the second chunk.
            self.assertEqual(next(result), second_chunk)
            # At last, no more.
            with self.assertRaises(StopIteration):
                next(result)


class DryRunTest(cros_test_lib.RunCommandTestCase):
    """Verify dry_run works for all of GSContext."""

    def setUp(self) -> None:
        self.ctx = gs.GSContext(dry_run=True)

    def tearDown(self) -> None:
        # Verify we don't try to call gsutil at all.
        for call_args in self.rc.call_args_list:
            self.assertNotIn("gsutil", call_args[0][0])

    def testCat(self) -> None:
        """Test Cat in dry_run mode."""
        self.assertEqual(self.ctx.Cat("gs://foo/bar"), b"")
        self.assertEqual(self.ctx.Cat("gs://foo/bar", encoding="utf-8"), "")

    def testChangeACL(self) -> None:
        """Test ChangeACL in dry_run mode."""
        self.assertEqual(
            self.ctx.ChangeACL("gs://foo/bar", acl_args_file="/dev/null"), None
        )

    def testCopy(self) -> None:
        """Test Copy in dry_run mode."""
        self.ctx.Copy("/dev/null", "gs://foo/bar")
        self.ctx.Copy("gs://foo/bar", "/dev/null")

    def testCreateWithContents(self) -> None:
        """Test Copy in dry_run mode."""
        self.ctx.CreateWithContents("gs://foo/bar", "My Little Content(tm)")

    def testCopyInto(self) -> None:
        """Test CopyInto in dry_run mode."""
        self.ctx.CopyInto("/dev/null", "gs://foo/bar")

    def testDoCommand(self) -> None:
        """Test DoCommand in dry_run mode."""
        self.ctx.DoCommand(["a-bad-command"])

    def testExists(self) -> None:
        """Test Exists in dry_run mode."""
        self.assertEqual(self.ctx.Exists("gs://foo/bar"), True)

    def testGetGeneration(self) -> None:
        """Test GetGeneration in dry_run mode."""
        self.assertEqual(self.ctx.GetGeneration("gs://foo/bar"), (0, 0))

    def testGetSize(self) -> None:
        """Test GetSize in dry_run mode."""
        self.assertEqual(self.ctx.GetSize("gs://foo/bar"), 0)

    def testGetTrackerFilenames(self) -> None:
        """Test GetTrackerFilenames in dry_run mode."""
        self.ctx.GetTrackerFilenames("foo")

    def testLS(self) -> None:
        """Test LS in dry_run mode."""
        self.assertEqual(self.ctx.LS("gs://foo/bar"), [])

    def testList(self) -> None:
        """Test List in dry_run mode."""
        self.assertEqual(self.ctx.List("gs://foo/bar"), [])

    def testMove(self) -> None:
        """Test Move in dry_run mode."""
        self.ctx.Move("gs://foo/bar", "gs://foo/bar2")

    def testRemove(self) -> None:
        """Test Remove in dry_run mode."""
        self.ctx.Remove("gs://foo/bar")

    def testSetACL(self) -> None:
        """Test SetACL in dry_run mode."""
        self.assertEqual(self.ctx.SetACL("gs://foo/bar", "bad-acl"), None)

    def testStat(self) -> None:
        """Test Stat in dry_run mode."""
        result = self.ctx.Stat("gs://foo/bar")
        self.assertEqual(result.content_length, 0)
        self.assertNotEqual(result.creation_time, None)

    def testStreamingCat(self) -> None:
        """Test StreamingCat in dry_run mode."""
        result = self.ctx.StreamingCat("gs://foo/bar")
        self.assertEqual(next(result), "")
        with self.assertRaises(StopIteration):
            next(result)

    def testVersion(self) -> None:
        """Test gsutil_version in dry_run mode."""
        self.assertEqual(self.ctx.gsutil_version, gs.GSContext.GSUTIL_VERSION)


class InitBotoTest(AbstractGSContextTest):
    """Test boto file interactive initialization."""

    # pylint: disable=protected-access

    GS_LS_ERROR = """\
You are attempting to access protected data with no configured credentials.
Please see http://code.google.com/apis/storage/docs/signup.html for
details about activating the Google Cloud Storage service and then run the
"gsutil config" command to configure gsutil to use these credentials."""

    GS_LS_ERROR2 = """\
GSResponseError: status=400, code=MissingSecurityHeader, reason=Bad Request, \
detail=Authorization."""

    GS_LS_BENIGN = """\
"GSResponseError: status=400, code=MissingSecurityHeader, reason=Bad Request,
detail=A nonempty x-goog-project-id header is required for this request."""

    def setUp(self) -> None:
        self.boto_file = os.path.join(self.tempdir, "boto_file")
        self.ctx = gs.GSContext(boto_file=self.boto_file)
        self.auth_cmd = ["ls", gs.AUTHENTICATION_BUCKET]

    def testGSLsSkippableError(self) -> None:
        """Benign GS error."""
        self.gs_mock.AddCmdResult(
            self.auth_cmd, returncode=1, stderr=self.GS_LS_BENIGN
        )
        self.assertTrue(self.ctx._TestGSLs())

    def testGSLsAuthorizationError1(self) -> None:
        """GS authorization error 1."""
        self.gs_mock.AddCmdResult(
            self.auth_cmd, returncode=1, stderr=self.GS_LS_ERROR
        )
        self.assertFalse(self.ctx._TestGSLs())

    def testGSLsAuthorizationErrorNoStderrCapture(self) -> None:
        """GS authorization error when not capturing stderr"""
        self.gs_mock.AddCmdResult(self.auth_cmd, returncode=1, stderr="")
        self.assertFalse(self.ctx._TestGSLs(stderr=False))

    def testGSLsError2(self) -> None:
        """GS authorization error 2."""
        self.gs_mock.AddCmdResult(
            self.auth_cmd, returncode=1, stderr=self.GS_LS_ERROR2
        )
        self.assertFalse(self.ctx._TestGSLs())

    def _WriteBotoFile(self, contents, *_args, **_kwargs) -> None:
        osutils.WriteFile(self.ctx.boto_file, contents)

    def testInitGSLsFailButSuccess(self) -> None:
        """Invalid GS Config, but we config properly."""
        self.gs_mock.AddCmdResult(
            self.auth_cmd, returncode=1, stderr=self.GS_LS_ERROR
        )
        self.ctx._InitBoto()

    def _AddLsConfigResult(self, side_effect=None) -> None:
        self.gs_mock.AddCmdResult(
            self.auth_cmd, returncode=1, stderr=self.GS_LS_ERROR
        )
        self.gs_mock.AddCmdResult(
            ["config"], returncode=1, side_effect=side_effect
        )

    def testGSLsFailAndConfigError(self) -> None:
        """Invalid GS Config, and we fail to config."""
        self._AddLsConfigResult(
            side_effect=functools.partial(self._WriteBotoFile, "monkeys")
        )
        self.assertRaises(cros_build_lib.RunCommandError, self.ctx._InitBoto)

    def testGSLsFailAndEmptyConfigFile(self) -> None:
        """Invalid GS Config, and we raise error on empty config file."""
        self._AddLsConfigResult(
            side_effect=functools.partial(self._WriteBotoFile, "")
        )
        self.assertRaises(gs.GSContextException, self.ctx._InitBoto)


class GSCounterTest(AbstractGSContextTest):
    """Tests GSCounter functionality."""

    COUNTER_URI = "gs://foo/mock/counter"
    INITIAL_VALUE = 100

    def setUp(self) -> None:
        self.counter = gs.GSCounter(self.ctx, self.COUNTER_URI)
        self.cat_mock = self.PatchObject(self.ctx, "Cat")
        self.gen_mock = self.PatchObject(
            self.ctx, "GetGeneration", return_value=(1, 1)
        )
        self._SetCounter(self.INITIAL_VALUE)

    def _SetCounter(self, value) -> None:
        """Set the test counter to |value|."""
        self.cat_mock.return_value = str(value)

    def testGetInitial(self) -> None:
        """Test Get when the counter doesn't exist."""
        self.cat_mock.side_effect = gs.GSNoSuchKey
        self.assertEqual(self.counter.Get(), 0)

    def testGet(self) -> None:
        """Basic Get() test."""
        self.assertEqual(self.counter.Get(), self.INITIAL_VALUE)

    def testIncrement(self) -> None:
        """Basic Increment() test."""
        self.assertEqual(self.counter.Increment(), self.INITIAL_VALUE + 1)

    def testDecrement(self) -> None:
        """Basic Decrement() test."""
        self.assertEqual(self.counter.Decrement(), self.INITIAL_VALUE - 1)

    def testReset(self) -> None:
        """Basic Reset() test."""
        self.assertEqual(self.counter.Reset(), 0)

    def testStreakIncrement(self) -> None:
        """Basic StreakIncrement() test."""
        self._SetCounter(10)
        self.assertEqual(self.counter.StreakIncrement(), 11)

    def testStreakIncrementReset(self) -> None:
        """Test StreakIncrement() when the counter is negative."""
        self._SetCounter(-10)
        self.assertEqual(self.counter.StreakIncrement(), 1)

    def testStreakDecrement(self) -> None:
        """Basic StreakDecrement() test."""
        self._SetCounter(-10)
        self.assertEqual(self.counter.StreakDecrement(), -11)

    def testStreakDecrementReset(self) -> None:
        """Test StreakDecrement() when the counter is positive."""
        self._SetCounter(10)
        self.assertEqual(self.counter.StreakDecrement(), -1)


class UnmockedGSCounterTest(cros_test_lib.TestCase):
    """Tests GSCounter functionality w/out mocks."""

    @staticmethod
    @contextlib.contextmanager
    def _Counter():
        ctx = gs.GSContext()
        with gs.TemporaryURL("chromite.counter") as tempuri:
            yield gs.GSCounter(ctx, tempuri)

    @staticmethod
    def _SetCounter(counter, value) -> None:
        """Set the test counter to |value|."""
        counter.AtomicCounterOperation(value, lambda x: value)

    @cros_test_lib.pytestmark_network_test
    def testGetInitial(self) -> None:
        """Test Get when the counter doesn't exist."""
        with self._Counter() as counter:
            self.assertEqual(counter.Get(), 0)

    @cros_test_lib.pytestmark_network_test
    def testGet(self) -> None:
        """Basic Get() test."""
        with self._Counter() as counter:
            self._SetCounter(counter, 100)
            self.assertEqual(counter.Get(), 100)

    @cros_test_lib.pytestmark_network_test
    def testIncrement(self) -> None:
        """Basic Increment() test."""
        with self._Counter() as counter:
            self._SetCounter(counter, 100)
            self.assertEqual(counter.Increment(), 101)

    @cros_test_lib.pytestmark_network_test
    def testDecrement(self) -> None:
        """Basic Decrement() test."""
        with self._Counter() as counter:
            self._SetCounter(counter, 100)
            self.assertEqual(counter.Decrement(), 99)

    @cros_test_lib.pytestmark_network_test
    def testReset(self) -> None:
        """Basic Reset() test."""
        with self._Counter() as counter:
            self._SetCounter(counter, 100)
            self.assertEqual(counter.Reset(), 0)
            self.assertEqual(counter.Get(), 0)

    @cros_test_lib.pytestmark_network_test
    def testStreakIncrement(self) -> None:
        """Basic StreakIncrement() test."""
        with self._Counter() as counter:
            self._SetCounter(counter, 100)
            self.assertEqual(counter.StreakIncrement(), 101)

    @cros_test_lib.pytestmark_network_test
    def testStreakIncrementReset(self) -> None:
        """Test StreakIncrement() when the counter is negative."""
        with self._Counter() as counter:
            self._SetCounter(counter, -100)
            self.assertEqual(counter.StreakIncrement(), 1)

    @cros_test_lib.pytestmark_network_test
    def testStreakDecrement(self) -> None:
        """Basic StreakDecrement() test."""
        with self._Counter() as counter:
            self._SetCounter(counter, -100)
            self.assertEqual(counter.StreakDecrement(), -101)

    @cros_test_lib.pytestmark_network_test
    def testStreakDecrementReset(self) -> None:
        """Test StreakDecrement() when the counter is positive."""
        with self._Counter() as counter:
            self._SetCounter(counter, 100)
            self.assertEqual(counter.StreakDecrement(), -1)