| Index: tools/telemetry/third_party/gsutilz/gslib/tests/testcase/base.py
|
| diff --git a/tools/telemetry/third_party/gsutilz/gslib/tests/testcase/base.py b/tools/telemetry/third_party/gsutilz/gslib/tests/testcase/base.py
|
| deleted file mode 100644
|
| index 9df8aceb4c321a3638aafba2df0a416ac7380a6b..0000000000000000000000000000000000000000
|
| --- a/tools/telemetry/third_party/gsutilz/gslib/tests/testcase/base.py
|
| +++ /dev/null
|
| @@ -1,148 +0,0 @@
|
| -# -*- coding: utf-8 -*-
|
| -# Copyright 2013 Google Inc. All Rights Reserved.
|
| -#
|
| -# Licensed under the Apache License, Version 2.0 (the "License");
|
| -# you may not use this file except in compliance with the License.
|
| -# You may obtain a copy of the License at
|
| -#
|
| -# http://www.apache.org/licenses/LICENSE-2.0
|
| -#
|
| -# Unless required by applicable law or agreed to in writing, software
|
| -# distributed under the License is distributed on an "AS IS" BASIS,
|
| -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
| -# See the License for the specific language governing permissions and
|
| -# limitations under the License.
|
| -"""Base test case class for unit and integration tests."""
|
| -
|
| -from __future__ import absolute_import
|
| -
|
| -from functools import wraps
|
| -import os.path
|
| -import random
|
| -import shutil
|
| -import tempfile
|
| -
|
| -import boto
|
| -import gslib.tests.util as util
|
| -from gslib.tests.util import unittest
|
| -from gslib.util import UTF8
|
| -
|
| -MAX_BUCKET_LENGTH = 63
|
| -
|
| -
|
| -def NotParallelizable(func):
|
| - """Wrapper function for cases that are not parallelizable."""
|
| - @wraps(func)
|
| - def ParallelAnnotatedFunc(*args, **kwargs):
|
| - return func(*args, **kwargs)
|
| - ParallelAnnotatedFunc.is_parallelizable = False
|
| - return ParallelAnnotatedFunc
|
| -
|
| -
|
| -def RequiresIsolation(func):
|
| - """Wrapper function for cases that require running in a separate process."""
|
| - @wraps(func)
|
| - def RequiresIsolationFunc(*args, **kwargs):
|
| - return func(*args, **kwargs)
|
| - RequiresIsolationFunc.requires_isolation = True
|
| - return RequiresIsolationFunc
|
| -
|
| -
|
| -class GsUtilTestCase(unittest.TestCase):
|
| - """Base test case class for unit and integration tests."""
|
| -
|
| - def setUp(self):
|
| - if util.RUN_S3_TESTS:
|
| - self.test_api = 'XML'
|
| - self.default_provider = 's3'
|
| - self.provider_custom_meta = 'amz'
|
| - else:
|
| - self.test_api = boto.config.get('GSUtil', 'prefer_api', 'JSON').upper()
|
| - self.default_provider = 'gs'
|
| - self.provider_custom_meta = 'goog'
|
| - self.tempdirs = []
|
| -
|
| - def tearDown(self):
|
| - while self.tempdirs:
|
| - tmpdir = self.tempdirs.pop()
|
| - shutil.rmtree(tmpdir, ignore_errors=True)
|
| -
|
| - def assertNumLines(self, text, numlines):
|
| - self.assertEqual(text.count('\n'), numlines)
|
| -
|
| - def GetTestMethodName(self):
|
| - if isinstance(self._testMethodName, unicode):
|
| - return self._testMethodName.encode(UTF8)
|
| - return self._testMethodName
|
| -
|
| - def MakeRandomTestString(self):
|
| - """Creates a random string of hex characters 8 characters long."""
|
| - return '%08x' % random.randrange(256**4)
|
| -
|
| - def MakeTempName(self, kind, prefix=''):
|
| - """Creates a temporary name that is most-likely unique.
|
| -
|
| - Args:
|
| - kind: A string indicating what kind of test name this is.
|
| - prefix: Prefix string to be used in the temporary name.
|
| -
|
| - Returns:
|
| - The temporary name.
|
| - """
|
| - name = '%sgsutil-test-%s-%s' % (prefix, self.GetTestMethodName(), kind)
|
| - name = name[:MAX_BUCKET_LENGTH-9]
|
| - name = '%s-%s' % (name, self.MakeRandomTestString())
|
| - return name
|
| -
|
| - def CreateTempDir(self, test_files=0):
|
| - """Creates a temporary directory on disk.
|
| -
|
| - The directory and all of its contents will be deleted after the test.
|
| -
|
| - Args:
|
| - test_files: The number of test files to place in the directory or a list
|
| - of test file names.
|
| -
|
| - Returns:
|
| - The path to the new temporary directory.
|
| - """
|
| - tmpdir = tempfile.mkdtemp(prefix=self.MakeTempName('directory'))
|
| - self.tempdirs.append(tmpdir)
|
| - try:
|
| - iter(test_files)
|
| - except TypeError:
|
| - test_files = [self.MakeTempName('file') for _ in range(test_files)]
|
| - for i, name in enumerate(test_files):
|
| - self.CreateTempFile(tmpdir=tmpdir, file_name=name, contents='test %d' % i)
|
| - return tmpdir
|
| -
|
| - def CreateTempFile(self, tmpdir=None, contents=None, file_name=None):
|
| - """Creates a temporary file on disk.
|
| -
|
| - Args:
|
| - tmpdir: The temporary directory to place the file in. If not specified, a
|
| - new temporary directory is created.
|
| - contents: The contents to write to the file. If not specified, a test
|
| - string is constructed and written to the file.
|
| - file_name: The name to use for the file. If not specified, a temporary
|
| - test file name is constructed. This can also be a tuple, where
|
| - ('dir', 'foo') means to create a file named 'foo' inside a
|
| - subdirectory named 'dir'.
|
| -
|
| - Returns:
|
| - The path to the new temporary file.
|
| - """
|
| - tmpdir = tmpdir or self.CreateTempDir()
|
| - file_name = file_name or self.MakeTempName('file')
|
| - if isinstance(file_name, basestring):
|
| - fpath = os.path.join(tmpdir, file_name)
|
| - else:
|
| - fpath = os.path.join(tmpdir, *file_name)
|
| - if not os.path.isdir(os.path.dirname(fpath)):
|
| - os.makedirs(os.path.dirname(fpath))
|
| -
|
| - with open(fpath, 'wb') as f:
|
| - contents = (contents if contents is not None
|
| - else self.MakeTempName('contents'))
|
| - f.write(contents)
|
| - return fpath
|
|
|