Revert "Add gsutil 4.13 to telemetry/third_party"

tools/telemetry/third_party/gsutil/gslib/tests/test_update.py

-# -*- coding: utf-8 -*-
-# Copyright 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.
-"""Tests for the update command."""
-
-from __future__ import absolute_import
-
-import os.path
-import shutil
-import subprocess
-import sys
-import tarfile
-
-import gslib
-import gslib.tests.testcase as testcase
-from gslib.tests.util import ObjectToURI as suri
-from gslib.tests.util import unittest
-from gslib.util import CERTIFICATE_VALIDATION_ENABLED
-
-
-TESTS_DIR = os.path.abspath(os.path.dirname(__file__))
-GSUTIL_DIR = os.path.join(TESTS_DIR, '..', '..')
-
-
-class UpdateTest(testcase.GsUtilIntegrationTestCase):
-  """Update command test suite."""
-
-  @unittest.skipUnless(CERTIFICATE_VALIDATION_ENABLED,
-                       'Test requires https certificate validation enabled.')
-  def test_update(self):
-    """Tests that the update command works or raises proper exceptions."""
-    if os.environ.get('CLOUDSDK_WRAPPER') == '1':
-      stderr = self.RunGsUtil(['update'], stdin='n',
-                              return_stderr=True, expected_status=1)
-      self.assertIn('update command is disabled for Cloud SDK', stderr)
-      return
-
-    if gslib.IS_PACKAGE_INSTALL:
-      # The update command is not present when installed via package manager.
-      stderr = self.RunGsUtil(['update'], return_stderr=True, expected_status=1)
-      self.assertIn('Invalid command', stderr)
-      return
-
-    # Create two temp directories, one of which we will run 'gsutil update' in
-    # to pull the changes from the other.
-    tmpdir_src = self.CreateTempDir()
-    tmpdir_dst = self.CreateTempDir()
-
-    # Copy gsutil to both source and destination directories.
-    gsutil_src = os.path.join(tmpdir_src, 'gsutil')
-    gsutil_dst = os.path.join(tmpdir_dst, 'gsutil')
-    # Path when executing from tmpdir (Windows doesn't support in-place rename)
-    gsutil_relative_dst = os.path.join('gsutil', 'gsutil')
-
-    shutil.copytree(GSUTIL_DIR, gsutil_src)
-    # Copy specific files rather than all of GSUTIL_DIR so we don't pick up temp
-    # working files left in top-level directory by gsutil developers (like tags,
-    # .git*, etc.)
-    os.makedirs(gsutil_dst)
-    for comp in ('CHANGES.md', 'CHECKSUM', 'COPYING', 'gslib', 'gsutil',
-                 'gsutil.py', 'MANIFEST.in', 'README.md', 'setup.py',
-                 'third_party', 'VERSION'):
-      if os.path.isdir(os.path.join(GSUTIL_DIR, comp)):
-        func = shutil.copytree
-      else:
-        func = shutil.copyfile
-      func(os.path.join(GSUTIL_DIR, comp), os.path.join(gsutil_dst, comp))
-
-    # Create a fake version number in the source so we can verify it in the
-    # destination.
-    expected_version = '17.25'
-    src_version_file = os.path.join(gsutil_src, 'VERSION')
-    self.assertTrue(os.path.exists(src_version_file))
-    with open(src_version_file, 'w') as f:
-      f.write(expected_version)
-
-    # Create a tarball out of the source directory and copy it to a bucket.
-    src_tarball = os.path.join(tmpdir_src, 'gsutil.test.tar.gz')
-
-    normpath = os.path.normpath
-    try:
-      # We monkey patch os.path.normpath here because the tarfile module
-      # normalizes the ./gsutil path, but the update command expects the tar
-      # file to be prefixed with . This preserves the ./gsutil path.
-      os.path.normpath = lambda fname: fname
-      tar = tarfile.open(src_tarball, 'w:gz')
-      tar.add(gsutil_src, arcname='./gsutil')
-      tar.close()
-    finally:
-      os.path.normpath = normpath
-
-    prefix = [sys.executable] if sys.executable else []
-
-    # Run with an invalid gs:// URI.
-    p = subprocess.Popen(prefix + ['gsutil', 'update', 'gs://pub'],
-                         cwd=gsutil_dst, stdout=subprocess.PIPE,
-                         stderr=subprocess.PIPE)
-    (_, stderr) = p.communicate()
-    p.stdout.close()
-    p.stderr.close()
-    self.assertEqual(p.returncode, 1)
-    self.assertIn('update command only works with tar.gz', stderr)
-
-    # Run with non-existent gs:// URI.
-    p = subprocess.Popen(
-        prefix + ['gsutil', 'update', 'gs://pub/Jdjh38)(;.tar.gz'],
-        cwd=gsutil_dst, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    (_, stderr) = p.communicate()
-    p.stdout.close()
-    p.stderr.close()
-    self.assertEqual(p.returncode, 1)
-    self.assertIn('NotFoundException', stderr)
-
-    # Run with file:// URI wihout -f option.
-    p = subprocess.Popen(prefix + ['gsutil', 'update', suri(src_tarball)],
-                         cwd=gsutil_dst, stdout=subprocess.PIPE,
-                         stderr=subprocess.PIPE)
-    (_, stderr) = p.communicate()
-    p.stdout.close()
-    p.stderr.close()
-    self.assertEqual(p.returncode, 1)
-    self.assertIn('command does not support', stderr)
-
-    # Run with a file present that was not distributed with gsutil.
-    with open(os.path.join(gsutil_dst, 'userdata.txt'), 'w') as fp:
-      fp.write('important data\n')
-    p = subprocess.Popen(prefix + ['gsutil', 'update', '-f', suri(src_tarball)],
-                         cwd=gsutil_dst, stdout=subprocess.PIPE,
-                         stderr=subprocess.PIPE, stdin=subprocess.PIPE)
-    (_, stderr) = p.communicate()
-    p.stdout.close()
-    p.stderr.close()
-    # Clean up before next test, and before assertions so failure doesn't leave
-    # this file around.
-    os.unlink(os.path.join(gsutil_dst, 'userdata.txt'))
-    self.assertEqual(p.returncode, 1)
-    self.assertIn(
-        'The update command cannot run with user data in the gsutil directory',
-        stderr.replace(os.linesep, ' '))
-
-    # Now do the real update, which should succeed.
-    p = subprocess.Popen(prefix + [gsutil_relative_dst, 'update', '-f',
-                                   suri(src_tarball)],
-                         cwd=tmpdir_dst, stdout=subprocess.PIPE,
-                         stderr=subprocess.PIPE, stdin=subprocess.PIPE)
-    (_, stderr) = p.communicate(input='y\r\n')
-    p.stdout.close()
-    p.stderr.close()
-    self.assertEqual(p.returncode, 0, msg=(
-        'Non-zero return code (%d) from gsutil update. stderr = \n%s' %
-        (p.returncode, stderr)))
-
-    # Verify that version file was updated.
-    dst_version_file = os.path.join(tmpdir_dst, 'gsutil', 'VERSION')
-    with open(dst_version_file, 'r') as f:
-      self.assertEqual(f.read(), expected_version)