Reland 0c7d94eb9d9fa388bda0d74405c0928dceea22ff: Add support for tar.gz archive files to download f…

tests/upload_to_google_storage_unittests.py

 #!/usr/bin/env python
 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Unit tests for upload_to_google_storage.py."""
 
 import optparse
 import os
 import Queue
 import shutil
 import StringIO
 import sys
+import tarfile
 import tempfile
 import threading
 import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 import upload_to_google_storage
 from download_from_google_storage_unittests import GsutilMock
+from download_from_google_storage_unittests import ChangedWorkingDirectory
 
 # ../third_party/gsutil/gsutil
 GSUTIL_DEFAULT_PATH = os.path.join(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
     'third_party', 'gsutil', 'gsutil')
 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
 
 
 class UploadTests(unittest.TestCase):
   def setUp(self):
@@ skipping 24 matching lines @@
          ('check_call',
           ('cp', '-z', 'txt', filenames[0],
            '%s/%s' % (self.base_url, self.lorem_ipsum_sha1)))])
     self.assertTrue(os.path.exists(output_filename))
     self.assertEqual(
         open(output_filename, 'rb').read(),
         '7871c8e24da15bad8b0be2c36edc9dc77e37727f')
     os.remove(output_filename)
     self.assertEqual(code, 0)
 
+  def test_create_archive(self):
+    work_dir = os.path.join(self.base_path, 'download_test_data')
+    with ChangedWorkingDirectory(work_dir):
+      dirname = 'subfolder'
+      dirs = [dirname]
+      tar_gz_file = '%s.tar.gz' % dirname
+      self.assertTrue(upload_to_google_storage.validate_archive_dirs(dirs))
+      upload_to_google_storage.create_archives(dirs)
+      self.assertTrue(os.path.exists(tar_gz_file))
+      with tarfile.open(tar_gz_file, 'r:gz') as tar:
+        content = map(lambda x: x.name, tar.getmembers())
+        self.assertTrue(dirname in content)
+        self.assertTrue(os.path.join(dirname, 'subfolder_text.txt') in content)
+        self.assertTrue(
+            os.path.join(dirname, 'subfolder_text.txt.sha1') in content)
+
+  def test_validate_archive_dirs_fails(self):
+    work_dir = os.path.join(self.base_path, 'download_test_data')
+    with ChangedWorkingDirectory(work_dir):
+      symlink = 'link'
+      os.symlink(os.path.join(self.base_path, 'subfolder'), symlink)
+    self.assertFalse(upload_to_google_storage.validate_archive_dirs([symlink]))
+    self.assertFalse(upload_to_google_storage.validate_archive_dirs(['foobar']))
+
   def test_upload_single_file_remote_exists(self):
     filenames = [self.lorem_ipsum]
     output_filename = '%s.sha1'  % self.lorem_ipsum
     etag_string = 'ETag: 634d7c1ed3545383837428f031840a1e'
     self.gsutil.add_expected(0, '', '')
     self.gsutil.add_expected(0, etag_string, '')
     code = upload_to_google_storage.upload_to_google_storage(
         filenames, self.base_url, self.gsutil, False, False, 1, False, None)
     self.assertEqual(
         self.gsutil.history,
@@ skipping 81 matching lines @@
     sys.stdin = StringIO.StringIO('\0'.join(inputs))
     result = upload_to_google_storage.get_targets(
         ['-'],
         self.parser,
         True)
     self.assertEqual(result, inputs)
 
 
 if __name__ == '__main__':
   unittest.main()