Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'dest': {'path': '', 'type': 'local'},
'dir_op': True, 'use_src_name': True}
params = {'region': 'us-east-1'}
files = FileGenerator(self.client, '').call(input_s3_file)
self.parsed_responses = [{
"CommonPrefixes": [], "Contents": [
{"Key": "another_directory/text2.txt", "Size": 100,
"LastModified": "2014-01-09T20:45:49.000Z"},
{"Key": "text1.txt", "Size": 10,
"LastModified": "2013-01-09T20:45:49.000Z"}]}]
self.patch_make_request()
result_list = []
for filename in files:
result_list.append(filename)
file_stat = FileStat(src=self.file2,
dest='another_directory' + os.sep +
'text2.txt',
compare_key='another_directory/text2.txt',
size=result_list[0].size,
last_update=result_list[0].last_update,
src_type='s3',
dest_type='local', operation_name='')
file_stat2 = FileStat(src=self.file1,
dest='text1.txt',
compare_key='text1.txt',
size=result_list[1].size,
last_update=result_list[1].last_update,
src_type='s3',
dest_type='local', operation_name='')
ref_list = [file_stat, file_stat2]
def test_compare_size_only_src_older_than_dest(self):
"""
Confirm that files with the same size but different update times are not
synced when `size_only` is set.
"""
time_dst = datetime.datetime.now()
time_src = time_dst + datetime.timedelta(days=1)
src_file = FileStat(src='', dest='',
compare_key='test.py', size=10,
last_update=time_src, src_type='local',
dest_type='s3', operation_name='upload')
dst_file = FileStat(src='', dest='',
compare_key='test.py', size=10,
last_update=time_dst, src_type='s3',
dest_type='local', operation_name='')
files = self.comparator.call(iter([src_file]), iter([dst_file]))
self.assertEqual(sum(1 for _ in files), 0)
src=self.bucket + '/another_directory/',
dest='another_directory' + os.sep,
compare_key='another_directory/',
size=0,
last_update=result_list[0].last_update,
src_type='s3',
dest_type='local', operation_name='delete')
file_stat2 = FileStat(
src=self.file2,
dest='another_directory' + os.sep + 'text2.txt',
compare_key='another_directory/text2.txt',
size=21,
last_update=result_list[1].last_update,
src_type='s3',
dest_type='local', operation_name='delete')
file_stat3 = FileStat(
src=self.file1,
dest='text1.txt',
compare_key='text1.txt',
size=15,
last_update=result_list[2].last_update,
src_type='s3',
dest_type='local', operation_name='delete')
expected_list = [file_stat1, file_stat2, file_stat3]
self.assertEqual(len(result_list), 3)
compare_files(self, result_list[0], expected_list[0])
compare_files(self, result_list[1], expected_list[1])
compare_files(self, result_list[2], expected_list[2])
def test_compare_lastmod_copy(self):
"""
Confirms compare time works for copies
"""
src_files = []
dest_files = []
ref_list = []
result_list = []
time = datetime.datetime.now()
future_time = time + datetime.timedelta(0, 3)
src_file = FileStat(src='', dest='',
compare_key='comparator_test.py', size=10,
last_update=future_time, src_type='s3',
dest_type='s3', operation_name='copy')
dest_file = FileStat(src='', dest='',
compare_key='comparator_test.py', size=10,
last_update=time, src_type='s3',
dest_type='s3', operation_name='')
src_files.append(src_file)
dest_files.append(dest_file)
files = self.comparator.call(iter(src_files), iter(dest_files))
ref_list.append(src_file)
for filename in files:
result_list.append(filename)
self.assertEqual(result_list, ref_list)
def test_local_directory(self):
"""
Generate an entire local directory.
"""
input_local_dir = {'src': {'path': self.local_dir,
'type': 'local'},
'dest': {'path': 'bucket/',
'type': 's3'},
'dir_op': True, 'use_src_name': True}
params = {'region': 'us-east-1'}
files = FileGenerator(self.client, '').call(input_local_dir)
result_list = []
for filename in files:
result_list.append(filename)
size, last_update = get_file_stat(self.local_file)
file_stat = FileStat(src=self.local_file, dest='bucket/text1.txt',
compare_key='text1.txt', size=size,
last_update=last_update, src_type='local',
dest_type='s3', operation_name='')
path = self.local_dir + 'another_directory' + os.sep \
+ 'text2.txt'
size, last_update = get_file_stat(path)
file_stat2 = FileStat(src=path,
dest='bucket/another_directory/text2.txt',
compare_key='another_directory/text2.txt',
size=size, last_update=last_update,
src_type='local',
dest_type='s3', operation_name='')
ref_list = [file_stat2, file_stat]
self.assertEqual(len(result_list), len(ref_list))
for i in range(len(result_list)):
compare_files(self, result_list[i], ref_list[i])
def test_compare_exact_timestamps_diff_age_not_download(self):
"""
Confirm that same sized files are synced when the timestamps differ,
the type of operation is not a download, and ``exact_timestamps``
is set.
"""
time_src = datetime.datetime.now()
time_dst = time_src - datetime.timedelta(days=1)
src_file = FileStat(src='', dest='',
compare_key='test.py', size=10,
last_update=time_src, src_type='s3',
dest_type='local', operation_name='upload')
dst_file = FileStat(src='', dest='',
compare_key='test.py', size=10,
last_update=time_dst, src_type='local',
dest_type='s3', operation_name='')
should_sync = self.sync_strategy.determine_should_sync(
src_file, dst_file)
self.assertTrue(should_sync)
def test_compare_size(self):
"""
Confirms compare size works.
"""
time = datetime.datetime.now()
src_file = FileStat(src='', dest='',
compare_key='comparator_test.py', size=11,
last_update=time, src_type='local',
dest_type='s3', operation_name='upload')
dest_file = FileStat(src='', dest='',
compare_key='comparator_test.py', size=10,
last_update=time, src_type='s3',
dest_type='local', operation_name='')
should_sync = self.sync_strategy.determine_should_sync(
src_file, dest_file)
self.assertTrue(should_sync)
def test_compare_exact_timestamps_same_age_same_size(self):
"""
Confirm that same-sized files are not synced when
the source and destination are the same age and
`exact_timestamps` is set.
"""
time_both = datetime.datetime.now()
src_file = FileStat(src='', dest='',
compare_key='test.py', size=10,
last_update=time_both, src_type='s3',
dest_type='local', operation_name='download')
dst_file = FileStat(src='', dest='',
compare_key='test.py', size=10,
last_update=time_both, src_type='local',
dest_type='s3', operation_name='')
files = self.comparator.call(iter([src_file]), iter([dst_file]))
self.assertEqual(sum(1 for _ in files), 0)
def test_compare_key_equal(self):
"""
Confirms checking compare key works.
"""
src_files = []
dest_files = []
ref_list = []
result_list = []
time = datetime.datetime.now()
src_file = FileStat(src='', dest='',
compare_key='comparator_test.py', size=10,
last_update=time, src_type='local',
dest_type='s3', operation_name='upload')
dest_file = FileStat(src='', dest='',
compare_key='comparator_test.py', size=10,
last_update=time, src_type='s3',
dest_type='local', operation_name='')
src_files.append(src_file)
dest_files.append(dest_file)
files = self.comparator.call(iter(src_files), iter(dest_files))
for filename in files:
result_list.append(filename)
self.assertEqual(result_list, ref_list)
def test_empty_dest(self):
"""
Confirm the appropriate action is taken when there are no more dest
files to take.
"""
src_files = []
dest_files = []
ref_list = []
result_list = []
time = datetime.datetime.now()
src_file = FileStat(src='', dest='',
compare_key='domparator_test.py', size=10,
last_update=time, src_type='local',
dest_type='s3', operation_name='upload')
src_files.append(src_file)
ref_list.append(src_file)
files = self.comparator.call(iter(src_files), iter(dest_files))
for filename in files:
result_list.append(filename)
self.assertEqual(result_list, ref_list)