Added gsutil to depot tools

third_party/gsutil/boto/tests/integration/gs/testcase.py

+# -*- coding: utf-8 -*-
+# Copyright (c) 2013, Google, Inc.
+# All rights reserved.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a
+# copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish, dis-
+# tribute, sublicense, and/or sell copies of the Software, and to permit
+# persons to whom the Software is furnished to do so, subject to the fol-
+# lowing conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+# IN THE SOFTWARE.
+
+"""Base TestCase class for gs integration tests."""
+
+import shutil
+import tempfile
+import time
+
+from boto.exception import GSResponseError
+from boto.gs.connection import GSConnection
+from tests.integration.gs.util import has_google_credentials
+from tests.integration.gs.util import retry
+from tests.unit import unittest
+
+@unittest.skipUnless(has_google_credentials(),
+                     "Google credentials are required to run the Google "
+                     "Cloud Storage tests.  Update your boto.cfg to run "
+                     "these tests.")
+class GSTestCase(unittest.TestCase):
+    gs = True
+
+    def setUp(self):
+        self._conn = GSConnection()
+        self._buckets = []
+        self._tempdirs = []
+
+    # Retry with an exponential backoff if a server error is received. This
+    # ensures that we try *really* hard to clean up after ourselves.
+    @retry(GSResponseError)
+    def tearDown(self):
+        while len(self._tempdirs):
+            tmpdir = self._tempdirs.pop()
+            shutil.rmtree(tmpdir, ignore_errors=True)
+
+        while(len(self._buckets)):
+            b = self._buckets[-1]
+            bucket = self._conn.get_bucket(b)
+            while len(list(bucket.list_versions())) > 0:
+                for k in bucket.list_versions():
+                    bucket.delete_key(k.name, generation=k.generation)
+            bucket.delete()
+            self._buckets.pop()
+
+    def _GetConnection(self):
+        """Returns the GSConnection object used to connect to GCS."""
+        return self._conn
+
+    def _MakeTempName(self):
+        """Creates and returns a temporary name for testing that is likely to be
+        unique."""
+        return "boto-gs-test-%s" % repr(time.time()).replace(".", "-")
+
+    def _MakeBucketName(self):
+        """Creates and returns a temporary bucket name for testing that is
+        likely to be unique."""
+        b = self._MakeTempName()
+        self._buckets.append(b)
+        return b
+
+    def _MakeBucket(self):
+        """Creates and returns temporary bucket for testing. After the test, the
+        contents of the bucket and the bucket itself will be deleted."""
+        b = self._conn.create_bucket(self._MakeBucketName())
+        return b
+
+    def _MakeKey(self, data='', bucket=None, set_contents=True):
+        """Creates and returns a Key with provided data. If no bucket is given,
+        a temporary bucket is created."""
+        if data and not set_contents:
+            # The data and set_contents parameters are mutually exclusive. 
+            raise ValueError('MakeKey called with a non-empty data parameter '
+                             'but set_contents was set to False.')
+        if not bucket:
+            bucket = self._MakeBucket()
+        key_name = self._MakeTempName()
+        k = bucket.new_key(key_name)
+        if set_contents:
+            k.set_contents_from_string(data)
+        return k
+
+    def _MakeVersionedBucket(self):
+        """Creates and returns temporary versioned bucket for testing. After the
+        test, the contents of the bucket and the bucket itself will be
+        deleted."""
+        b = self._MakeBucket()
+        b.configure_versioning(True)
+        return b
+
+    def _MakeTempDir(self):
+        """Creates and returns a temporary directory on disk. After the test,
+        the contents of the directory and the directory itself will be
+        deleted."""
+        tmpdir = tempfile.mkdtemp(prefix=self._MakeTempName())
+        self._tempdirs.append(tmpdir)
+        return tmpdir