add only_if_modified param to GSUtils.upload_file()

py/utils/gs_utils_manualtest.py

 #!/usr/bin/python
 
 """Tests for gs_utils.py.
 
 TODO(epoger): How should we exercise these self-tests? See http://skbug.com/2751
 """
 
 # System-level imports.
 import os
 import posixpath
 import random
 import shutil
 import sys
 import tempfile
+import time
 
 # Local imports.
 import gs_utils
 
 TEST_BUCKET = 'chromium-skia-testing'
 
 
 def _get_authenticated_gs_handle():
   """Returns an instance of GSUtils using ~/.boto for authentication."""
   try:
@@ skipping 12 matching lines @@
   """Returns a unique directory name suitable for use in Google Storage."""
   return 'gs_utils_manualtest/%d' % random.randint(0, sys.maxint)
 
 
 def _test_public_read():
   """Make sure we can read from public files without .boto file credentials."""
   gs = gs_utils.GSUtils()
   gs.list_bucket_contents(bucket=TEST_BUCKET, subdir=None)
 
 
+def _test_only_if_modified():
+  """Test only_if_modified param within upload_file()."""
+  gs = _get_authenticated_gs_handle()
+  filename = 'filename'
+  remote_dir = _get_unique_posix_dir()
+  dest_path = posixpath.join(remote_dir, filename)
+  local_dir = tempfile.mkdtemp()
+  try:
+    # Create a file on local disk, and upload it for the first time.
+    local_path = os.path.join(local_dir, filename)
+    with open(local_path, 'w') as f:
+      f.write('original contents')
+    gs.upload_file(source_path=local_path, dest_bucket=TEST_BUCKET,
+                   dest_path=dest_path, only_if_modified=True)
+    try:
+      # Re-upload the same file one second later, with only_if_modified=False;
+      # the timestamp should change.
+      old_timestamp = gs.get_last_modified_time(
+          bucket=TEST_BUCKET, path=dest_path)
+      time.sleep(2)
+      gs.upload_file(source_path=local_path, dest_bucket=TEST_BUCKET,
+                     dest_path=dest_path, only_if_modified=False)
+      new_timestamp = gs.get_last_modified_time(
+          bucket=TEST_BUCKET, path=dest_path)
+      assert old_timestamp != new_timestamp, '%s != %s' % (
+          old_timestamp, new_timestamp)
+
+      # Re-upload the same file one second later, with only_if_modified=True;
+      # the timestamp should NOT change.
+      old_timestamp = new_timestamp
+      time.sleep(2)
+      gs.upload_file(source_path=local_path, dest_bucket=TEST_BUCKET,
+                     dest_path=dest_path, only_if_modified=True)
+      new_timestamp = gs.get_last_modified_time(
+          bucket=TEST_BUCKET, path=dest_path)
+      assert old_timestamp == new_timestamp, '%s == %s' % (
+          old_timestamp, new_timestamp)
+
+      # MODIFY and re-upload the file one second later, with
+      # only_if_modified=True; the timestamp SHOULD change.
+      old_timestamp = new_timestamp
+      with open(local_path, 'w') as f:
+        f.write('modified contents')
+      time.sleep(2)
+      gs.upload_file(source_path=local_path, dest_bucket=TEST_BUCKET,
+                     dest_path=dest_path, only_if_modified=True)
+      new_timestamp = gs.get_last_modified_time(
+          bucket=TEST_BUCKET, path=dest_path)
+      assert old_timestamp != new_timestamp, '%s != %s' % (
+          old_timestamp, new_timestamp)
+    finally:
+      # Clean up the remote_dir.
+      gs.delete_file(bucket=TEST_BUCKET, path=dest_path)
+  finally:
+    # Clean up the local dir.
+    shutil.rmtree(local_dir)
+
 def _test_authenticated_round_trip():
   gs = _get_authenticated_gs_handle()
   remote_dir = _get_unique_posix_dir()
   subdir = 'subdir'
   filenames_to_upload = ['file1', 'file2']
 
   # Upload test files to Google Storage, checking that their fine-grained
   # ACLs were set correctly.
   id_type = gs.IdType.GROUP_BY_DOMAIN
   id_value = 'chromium.org'
@@ skipping 142 matching lines @@
       assert file_contents == 'contents of %s\n' % filename, (
           '%s == "contents of %s\n"' % (file_contents, filename))
   finally:
     shutil.rmtree(local_dest_dir)
     for filename in filenames:
       gs.delete_file(bucket=TEST_BUCKET,
                      path=posixpath.join(remote_dir, subdir, filename))
 
 
 if __name__ == '__main__':
+  _test_only_if_modified()
   _test_public_read()
   _test_authenticated_round_trip()
   _test_dir_upload_and_download()
   # TODO(epoger): Add _test_unauthenticated_access() to make sure we raise
   # an exception when we try to access without needed credentials.