tools/telemetry/third_party/gsutilz/gslib/tests/test_rsync_funcs.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright 2014 Google Inc. All Rights Reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 """Unit tests for functions in rsync command."""
  16
  17 import logging
  18 import os
  19
  20 from gslib.commands.rsync import _ComputeNeededFileChecksums
  21 from gslib.commands.rsync import _NA
  22 from gslib.hashing_helper import CalculateB64EncodedCrc32cFromContents
  23 from gslib.hashing_helper import CalculateB64EncodedMd5FromContents
  24 from gslib.tests.testcase.unit_testcase import GsUtilUnitTestCase
  25
  26
  27 class TestRsyncFuncs(GsUtilUnitTestCase):
  28
  29   def test_compute_needed_file_checksums(self):
  30     """Tests that we compute all/only needed file checksums."""
  31     size = 4
  32     logger = logging.getLogger()
  33     tmpdir = self.CreateTempDir()
  34     file_url_str = 'file://%s' % os.path.join(tmpdir, 'obj1')
  35     self.CreateTempFile(tmpdir=tmpdir, file_name='obj1', contents='obj1')
  36     cloud_url_str = 'gs://whatever'
  37     with open(os.path.join(tmpdir, 'obj1'), 'rb') as fp:
  38       crc32c = CalculateB64EncodedCrc32cFromContents(fp)
  39       fp.seek(0)
  40       md5 = CalculateB64EncodedMd5FromContents(fp)
  41
  42     # Test case where source is a file and dest has CRC32C.
  43     (src_crc32c, src_md5, dst_crc32c, dst_md5) = _ComputeNeededFileChecksums(
  44         logger, file_url_str, size, _NA, _NA, cloud_url_str, size, crc32c, _NA)
  45     self.assertEquals(crc32c, src_crc32c)
  46     self.assertEquals(_NA, src_md5)
  47     self.assertEquals(crc32c, dst_crc32c)
  48     self.assertEquals(_NA, dst_md5)
  49
  50     # Test case where source is a file and dest has MD5 but not CRC32C.
  51     (src_crc32c, src_md5, dst_crc32c, dst_md5) = _ComputeNeededFileChecksums(
  52         logger, file_url_str, size, _NA, _NA, cloud_url_str, size, _NA, md5)
  53     self.assertEquals(_NA, src_crc32c)
  54     self.assertEquals(md5, src_md5)
  55     self.assertEquals(_NA, dst_crc32c)
  56     self.assertEquals(md5, dst_md5)
  57
  58     # Test case where dest is a file and src has CRC32C.
  59     (src_crc32c, src_md5, dst_crc32c, dst_md5) = _ComputeNeededFileChecksums(
  60         logger, cloud_url_str, size, crc32c, _NA, file_url_str, size, _NA, _NA)
  61     self.assertEquals(crc32c, dst_crc32c)
  62     self.assertEquals(_NA, src_md5)
  63     self.assertEquals(crc32c, src_crc32c)
  64     self.assertEquals(_NA, src_md5)
  65
  66     # Test case where dest is a file and src has MD5 but not CRC32C.
  67     (src_crc32c, src_md5, dst_crc32c, dst_md5) = _ComputeNeededFileChecksums(
  68         logger, cloud_url_str, size, _NA, md5, file_url_str, size, _NA, _NA)
  69     self.assertEquals(_NA, dst_crc32c)
  70     self.assertEquals(md5, src_md5)
  71     self.assertEquals(_NA, src_crc32c)
  72     self.assertEquals(md5, src_md5)