OLD | NEW |
1 #!/usr/bin/env python | 1 #!/usr/bin/env python |
2 # Copyright (c) 2012 The Chromium Authors. All rights reserved. | 2 # Copyright (c) 2012 The Chromium Authors. All rights reserved. |
3 # Use of this source code is governed by a BSD-style license that can be | 3 # Use of this source code is governed by a BSD-style license that can be |
4 # found in the LICENSE file. | 4 # found in the LICENSE file. |
5 | 5 |
6 """Unit tests for upload_to_google_storage.py.""" | 6 """Unit tests for upload_to_google_storage.py.""" |
7 | 7 |
8 import optparse | 8 import optparse |
9 import os | 9 import os |
10 import Queue | 10 import Queue |
(...skipping 30 matching lines...) Expand all Loading... |
41 self.lorem_ipsum_sha1 = '7871c8e24da15bad8b0be2c36edc9dc77e37727f' | 41 self.lorem_ipsum_sha1 = '7871c8e24da15bad8b0be2c36edc9dc77e37727f' |
42 | 42 |
43 def cleanUp(self): | 43 def cleanUp(self): |
44 shutil.rmtree(self.temp_dir) | 44 shutil.rmtree(self.temp_dir) |
45 sys.stdin = sys.__stdin__ | 45 sys.stdin = sys.__stdin__ |
46 | 46 |
47 def test_upload_single_file(self): | 47 def test_upload_single_file(self): |
48 filenames = [self.lorem_ipsum] | 48 filenames = [self.lorem_ipsum] |
49 output_filename = '%s.sha1' % self.lorem_ipsum | 49 output_filename = '%s.sha1' % self.lorem_ipsum |
50 code = upload_to_google_storage.upload_to_google_storage( | 50 code = upload_to_google_storage.upload_to_google_storage( |
51 filenames, self.base_url, self.gsutil, True, False, 1, False) | 51 filenames, self.base_url, self.gsutil, True, False, 1, False, 'txt') |
52 self.assertEqual( | 52 self.assertEqual( |
53 self.gsutil.history, | 53 self.gsutil.history, |
54 [('check_call', | 54 [('check_call', |
55 ('ls', '%s/%s' % (self.base_url, self.lorem_ipsum_sha1))), | 55 ('ls', '%s/%s' % (self.base_url, self.lorem_ipsum_sha1))), |
56 ('check_call', | 56 ('check_call', |
57 ('cp', '-q', filenames[0], '%s/%s' % (self.base_url, | 57 ('cp', '-z', 'txt', filenames[0], |
58 self.lorem_ipsum_sha1)))]) | 58 '%s/%s' % (self.base_url, self.lorem_ipsum_sha1)))]) |
59 self.assertTrue(os.path.exists(output_filename)) | 59 self.assertTrue(os.path.exists(output_filename)) |
60 self.assertEqual( | 60 self.assertEqual( |
61 open(output_filename, 'rb').read(), | 61 open(output_filename, 'rb').read(), |
62 '7871c8e24da15bad8b0be2c36edc9dc77e37727f') | 62 '7871c8e24da15bad8b0be2c36edc9dc77e37727f') |
63 os.remove(output_filename) | 63 os.remove(output_filename) |
64 self.assertEqual(code, 0) | 64 self.assertEqual(code, 0) |
65 | 65 |
66 def test_upload_single_file_remote_exists(self): | 66 def test_upload_single_file_remote_exists(self): |
67 filenames = [self.lorem_ipsum] | 67 filenames = [self.lorem_ipsum] |
68 output_filename = '%s.sha1' % self.lorem_ipsum | 68 output_filename = '%s.sha1' % self.lorem_ipsum |
69 etag_string = 'ETag: 634d7c1ed3545383837428f031840a1e' | 69 etag_string = 'ETag: 634d7c1ed3545383837428f031840a1e' |
70 self.gsutil.add_expected(0, '', '') | 70 self.gsutil.add_expected(0, '', '') |
71 self.gsutil.add_expected(0, etag_string, '') | 71 self.gsutil.add_expected(0, etag_string, '') |
72 code = upload_to_google_storage.upload_to_google_storage( | 72 code = upload_to_google_storage.upload_to_google_storage( |
73 filenames, self.base_url, self.gsutil, False, False, 1, False) | 73 filenames, self.base_url, self.gsutil, False, False, 1, False, None) |
74 self.assertEqual( | 74 self.assertEqual( |
75 self.gsutil.history, | 75 self.gsutil.history, |
76 [('check_call', | 76 [('check_call', |
77 ('ls', '%s/%s' % (self.base_url, self.lorem_ipsum_sha1))), | 77 ('ls', '%s/%s' % (self.base_url, self.lorem_ipsum_sha1))), |
78 ('check_call', | 78 ('check_call', |
79 ('ls', '-L', '%s/%s' % (self.base_url, self.lorem_ipsum_sha1)))]) | 79 ('ls', '-L', '%s/%s' % (self.base_url, self.lorem_ipsum_sha1)))]) |
80 self.assertTrue(os.path.exists(output_filename)) | 80 self.assertTrue(os.path.exists(output_filename)) |
81 self.assertEqual( | 81 self.assertEqual( |
82 open(output_filename, 'rb').read(), | 82 open(output_filename, 'rb').read(), |
83 '7871c8e24da15bad8b0be2c36edc9dc77e37727f') | 83 '7871c8e24da15bad8b0be2c36edc9dc77e37727f') |
84 os.remove(output_filename) | 84 os.remove(output_filename) |
85 self.assertEqual(code, 0) | 85 self.assertEqual(code, 0) |
86 | 86 |
87 def test_upload_worker_errors(self): | 87 def test_upload_worker_errors(self): |
88 work_queue = Queue.Queue() | 88 work_queue = Queue.Queue() |
89 work_queue.put((self.lorem_ipsum, self.lorem_ipsum_sha1)) | 89 work_queue.put((self.lorem_ipsum, self.lorem_ipsum_sha1)) |
90 work_queue.put((None, None)) | 90 work_queue.put((None, None)) |
91 self.gsutil.add_expected(1, '', '') # For the first ls call. | 91 self.gsutil.add_expected(1, '', '') # For the first ls call. |
92 self.gsutil.add_expected(20, '', 'Expected error message') | 92 self.gsutil.add_expected(20, '', 'Expected error message') |
93 # pylint: disable=W0212 | 93 # pylint: disable=W0212 |
94 upload_to_google_storage._upload_worker( | 94 upload_to_google_storage._upload_worker( |
95 0, | 95 0, |
96 work_queue, | 96 work_queue, |
97 self.base_url, | 97 self.base_url, |
98 self.gsutil, | 98 self.gsutil, |
99 threading.Lock(), | 99 threading.Lock(), |
100 False, | 100 False, |
101 False, | 101 False, |
102 self.stdout_queue, | 102 self.stdout_queue, |
103 self.ret_codes) | 103 self.ret_codes, |
| 104 None) |
104 expected_ret_codes = [ | 105 expected_ret_codes = [ |
105 (20, | 106 (20, |
106 'Encountered error on uploading %s to %s/%s\nExpected error message' % | 107 'Encountered error on uploading %s to %s/%s\nExpected error message' % |
107 (self.lorem_ipsum, self.base_url, self.lorem_ipsum_sha1))] | 108 (self.lorem_ipsum, self.base_url, self.lorem_ipsum_sha1))] |
108 self.assertEqual(list(self.ret_codes.queue), expected_ret_codes) | 109 self.assertEqual(list(self.ret_codes.queue), expected_ret_codes) |
109 | 110 |
110 def test_skip_hashing(self): | 111 def test_skip_hashing(self): |
111 filenames = [self.lorem_ipsum] | 112 filenames = [self.lorem_ipsum] |
112 output_filename = '%s.sha1' % self.lorem_ipsum | 113 output_filename = '%s.sha1' % self.lorem_ipsum |
113 fake_hash = '6871c8e24da15bad8b0be2c36edc9dc77e37727f' | 114 fake_hash = '6871c8e24da15bad8b0be2c36edc9dc77e37727f' |
114 with open(output_filename, 'wb') as f: | 115 with open(output_filename, 'wb') as f: |
115 f.write(fake_hash) # Fake hash. | 116 f.write(fake_hash) # Fake hash. |
116 code = upload_to_google_storage.upload_to_google_storage( | 117 code = upload_to_google_storage.upload_to_google_storage( |
117 filenames, self.base_url, self.gsutil, False, False, 1, True) | 118 filenames, self.base_url, self.gsutil, False, False, 1, True, None) |
118 self.assertEqual( | 119 self.assertEqual( |
119 self.gsutil.history, | 120 self.gsutil.history, |
120 [('check_call', | 121 [('check_call', |
121 ('ls', '%s/%s' % (self.base_url, fake_hash))), | 122 ('ls', '%s/%s' % (self.base_url, fake_hash))), |
122 ('check_call', | 123 ('check_call', |
123 ('ls', '-L', '%s/%s' % (self.base_url, fake_hash))), | 124 ('ls', '-L', '%s/%s' % (self.base_url, fake_hash))), |
124 ('check_call', | 125 ('check_call', |
125 ('cp', '-q', filenames[0], '%s/%s' % (self.base_url, fake_hash)))]) | 126 ('cp', filenames[0], '%s/%s' % (self.base_url, fake_hash)))]) |
126 self.assertEqual( | 127 self.assertEqual( |
127 open(output_filename, 'rb').read(), fake_hash) | 128 open(output_filename, 'rb').read(), fake_hash) |
128 os.remove(output_filename) | 129 os.remove(output_filename) |
129 self.assertEqual(code, 0) | 130 self.assertEqual(code, 0) |
130 | 131 |
131 def test_get_targets_no_args(self): | 132 def test_get_targets_no_args(self): |
132 try: | 133 try: |
133 upload_to_google_storage.get_targets([], self.parser, False) | 134 upload_to_google_storage.get_targets([], self.parser, False) |
134 self.fail() | 135 self.fail() |
135 except SystemExit, e: | 136 except SystemExit, e: |
(...skipping 19 matching lines...) Expand all Loading... |
155 inputs = ['a', 'b', 'c', 'd', 'e'] | 156 inputs = ['a', 'b', 'c', 'd', 'e'] |
156 sys.stdin = StringIO.StringIO('\0'.join(inputs)) | 157 sys.stdin = StringIO.StringIO('\0'.join(inputs)) |
157 result = upload_to_google_storage.get_targets( | 158 result = upload_to_google_storage.get_targets( |
158 ['-'], | 159 ['-'], |
159 self.parser, | 160 self.parser, |
160 True) | 161 True) |
161 self.assertEqual(result, inputs) | 162 self.assertEqual(result, inputs) |
162 | 163 |
163 | 164 |
164 if __name__ == '__main__': | 165 if __name__ == '__main__': |
165 unittest.main() | 166 unittest.main() |
OLD | NEW |