CoCalc -- test_comparator.py

GitHub Repository: aws/aws-cli
Path: blob/develop/tests/unit/customizations/s3/test_comparator.py
¹⁵⁶⁹ views
1
# Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2
#
3
# Licensed under the Apache License, Version 2.0 (the "License"). You
4
# may not use this file except in compliance with the License. A copy of
5
# the License is located at
6
#
7
#     http://aws.amazon.com/apache2.0/
8
#
9
# or in the "license" file accompanying this file. This file is
10
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11
# ANY KIND, either express or implied. See the License for the specific
12
# language governing permissions and limitations under the License.
13
import datetime
14
import unittest
15

16
from awscli.testutils import mock
17

18
from awscli.customizations.s3.comparator import Comparator
19
from awscli.customizations.s3.filegenerator import FileStat
20

21

22
class ComparatorTest(unittest.TestCase):
23
    def setUp(self):
24
        self.sync_strategy = mock.Mock()
25
        self.not_at_src_sync_strategy = mock.Mock()
26
        self.not_at_dest_sync_strategy = mock.Mock()
27
        self.comparator = Comparator(self.sync_strategy,
28
                                     self.not_at_dest_sync_strategy,
29
                                     self.not_at_src_sync_strategy)
30

31
    def test_compare_key_equal_should_not_sync(self):
32
        """
33
        Confirm the appropriate action is taken when the soruce compare key
34
        is equal to the destination compare key.
35
        """
36
        # Try when the sync strategy says not to sync the file.
37
        self.sync_strategy.determine_should_sync.return_value = False
38

39
        src_files = []
40
        dest_files = []
41
        ref_list = []
42
        result_list = []
43
        time = datetime.datetime.now()
44
        src_file = FileStat(src='', dest='',
45
                            compare_key='comparator_test.py', size=10,
46
                            last_update=time, src_type='local',
47
                            dest_type='s3', operation_name='upload')
48
        dest_file = FileStat(src='', dest='',
49
                             compare_key='comparator_test.py', size=10,
50
                             last_update=time, src_type='s3',
51
                             dest_type='local', operation_name='')
52
        src_files.append(src_file)
53
        dest_files.append(dest_file)
54
        files = self.comparator.call(iter(src_files), iter(dest_files))
55
        for filename in files:
56
            result_list.append(filename)
57
        self.assertEqual(result_list, ref_list)
58

59
        # Try when the sync strategy says to sync the file.
60
        self.sync_strategy.determine_should_sync.return_value = True
61

62
        ref_list = []
63
        result_list = []
64
        files = self.comparator.call(iter(src_files), iter(dest_files))
65
        ref_list.append(src_file)
66
        for filename in files:
67
            result_list.append(filename)
68
        self.assertEqual(result_list, ref_list)
69

70
    def test_compare_key_less(self):
71
        """
72
        Confirm the appropriate action is taken when the soruce compare key
73
        is less than the destination compare key.
74
        """
75
        self.not_at_src_sync_strategy.determine_should_sync.return_value = False
76

77
        # Try when the sync strategy says to sync the file.
78
        self.not_at_dest_sync_strategy.determine_should_sync.return_value = True
79

80
        src_files = []
81
        dest_files = []
82
        ref_list = []
83
        result_list = []
84
        time = datetime.datetime.now()
85
        src_file = FileStat(src='', dest='',
86
                            compare_key='bomparator_test.py', size=10,
87
                            last_update=time, src_type='local',
88
                            dest_type='s3', operation_name='upload')
89
        dest_file = FileStat(src='', dest='',
90
                             compare_key='comparator_test.py', size=10,
91
                             last_update=time, src_type='s3',
92
                             dest_type='local', operation_name='')
93
        src_files.append(src_file)
94
        dest_files.append(dest_file)
95
        ref_list.append(src_file)
96
        files = self.comparator.call(iter(src_files), iter(dest_files))
97
        for filename in files:
98
            result_list.append(filename)
99
        self.assertEqual(result_list, ref_list)
100

101
        # Now try when the sync strategy says not to sync the file.
102
        self.not_at_dest_sync_strategy.determine_should_sync.return_value = False
103
        result_list = []
104
        ref_list = []
105
        files = self.comparator.call(iter(src_files), iter(dest_files))
106
        for filename in files:
107
            result_list.append(filename)
108
        self.assertEqual(result_list, ref_list)
109

110

111
    def test_compare_key_greater(self):
112
        """
113
        Confirm the appropriate action is taken when the soruce compare key
114
        is greater than the destination compare key.
115
        """
116
        self.not_at_dest_sync_strategy.determine_should_sync.return_value = False
117

118
        # Try when the sync strategy says to sync the file.
119
        self.not_at_src_sync_strategy.determine_should_sync.return_value = True
120

121
        src_files = []
122
        dest_files = []
123
        ref_list = []
124
        result_list = []
125
        time = datetime.datetime.now()
126
        src_file = FileStat(src='', dest='',
127
                            compare_key='domparator_test.py', size=10,
128
                            last_update=time, src_type='local',
129
                            dest_type='s3', operation_name='upload')
130
        dest_file = FileStat(src='', dest='',
131
                             compare_key='comparator_test.py', size=10,
132
                             last_update=time, src_type='s3',
133
                             dest_type='local', operation_name='')
134
        src_files.append(src_file)
135
        dest_files.append(dest_file)
136
        ref_list.append(dest_file)
137
        files = self.comparator.call(iter(src_files), iter(dest_files))
138
        for filename in files:
139
            result_list.append(filename)
140
        self.assertEqual(result_list, ref_list)
141

142
        # Now try when the sync strategy says not to sync the file.
143
        self.not_at_src_sync_strategy.determine_should_sync.return_value = False
144
        result_list = []
145
        ref_list = []
146
        files = self.comparator.call(iter(src_files), iter(dest_files))
147
        for filename in files:
148
            result_list.append(filename)
149
        self.assertEqual(result_list, ref_list)
150

151

152
    def test_empty_src(self):
153
        """
154
        Confirm the appropriate action is taken when there are no more source
155
        files to take.
156
        """
157
        # Try when the sync strategy says to sync the file.
158
        self.not_at_src_sync_strategy.determine_should_sync.return_value = True
159

160
        src_files = []
161
        dest_files = []
162
        ref_list = []
163
        result_list = []
164
        time = datetime.datetime.now()
165
        dest_file = FileStat(src='', dest='',
166
                             compare_key='comparator_test.py', size=10,
167
                             last_update=time, src_type='s3',
168
                             dest_type='local', operation_name='')
169
        dest_files.append(dest_file)
170
        ref_list.append(dest_file)
171
        files = self.comparator.call(iter(src_files), iter(dest_files))
172
        for filename in files:
173
            result_list.append(filename)
174
        self.assertEqual(result_list, ref_list)
175

176
        # Now try when the sync strategy says not to sync the file.
177
        self.not_at_src_sync_strategy.determine_should_sync.return_value = False
178
        result_list = []
179
        ref_list = []
180
        files = self.comparator.call(iter(src_files), iter(dest_files))
181
        for filename in files:
182
            result_list.append(filename)
183
        self.assertEqual(result_list, ref_list)
184

185
    def test_empty_dest(self):
186
        """
187
        Confirm the appropriate action is taken when there are no more dest
188
        files to take.
189
        """
190
        # Try when the sync strategy says to sync the file.
191
        self.not_at_dest_sync_strategy.determine_should_sync.return_value = True
192

193
        src_files = []
194
        dest_files = []
195
        ref_list = []
196
        result_list = []
197
        time = datetime.datetime.now()
198
        src_file = FileStat(src='', dest='',
199
                            compare_key='domparator_test.py', size=10,
200
                            last_update=time, src_type='local',
201
                            dest_type='s3', operation_name='upload')
202
        src_files.append(src_file)
203
        ref_list.append(src_file)
204
        files = self.comparator.call(iter(src_files), iter(dest_files))
205
        for filename in files:
206
            result_list.append(filename)
207
        self.assertEqual(result_list, ref_list)
208

209
        # Now try when the sync strategy says not to sync the file.
210
        self.not_at_dest_sync_strategy.determine_should_sync.return_value = False
211
        result_list = []
212
        ref_list = []
213
        files = self.comparator.call(iter(src_files), iter(dest_files))
214
        for filename in files:
215
            result_list.append(filename)
216
        self.assertEqual(result_list, ref_list)
217

218

219
    def test_empty_src_dest(self):
220
        """
221
        Confirm the appropriate action is taken when there are no more
222
        files to take for both source and destination.
223
        """
224
        src_files = []
225
        dest_files = []
226
        ref_list = []
227
        result_list = []
228
        files = self.comparator.call(iter(src_files), iter(dest_files))
229
        for filename in files:
230
            result_list.append(filename)
231
        self.assertEqual(result_list, ref_list)
232

233

234
if __name__ == "__main__":
235
    unittest.main()
236

237
Product

Resources

Company
