Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(80)

Side by Side Diff: slave/skia_slave_scripts/utils/old_gs_utils_test.py

Issue 648353002: Remove Skia's forked buildbot code (Closed) Base URL: https://skia.googlesource.com/buildbot.git@master
Patch Set: Fix launch_slaves, remove more stuff Created 6 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
(Empty)
1 #!/usr/bin/env python
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
4 # found in the LICENSE file.
5
6 """Tests for module old_gs_utils."""
7
8 import __builtin__
9 import os
10 import posixpath
11 import shutil
12 import sys
13 import tempfile
14 import time
15
16 # Appending to PYTHONPATH to find common.
17 BUILDBOT_PATH = os.path.realpath(os.path.join(
18 os.path.dirname(os.path.abspath(__file__)), os.pardir, os.pardir, os.pardir)
19 )
20 sys.path.append(os.path.join(BUILDBOT_PATH, 'common'))
21 sys.path.append(os.path.join(BUILDBOT_PATH, 'third_party', 'chromium_buildbot',
22 'scripts'))
23 sys.path.append(os.path.join(BUILDBOT_PATH, 'third_party', 'chromium_buildbot',
24 'scripts', 'common'))
25 sys.path.append(os.path.join(BUILDBOT_PATH, 'third_party', 'chromium_buildbot',
26 'site_config'))
27 sys.path.append(os.path.join(BUILDBOT_PATH, 'third_party', 'chromium_buildbot',
28 'third_party', 'twisted_10_2'))
29
30 from py.utils import shell_utils
31 from slave import slave_utils
32 import old_gs_utils as gs_utils
33 import unittest
34
35
36 GSUTIL_LOCATION = slave_utils.GSUtilSetup()
37
38 TEST_TIMESTAMP = '1354128965'
39 TEST_TIMESTAMP_2 = '1354128985'
40
41
42 class TestGSUtils(unittest.TestCase):
43
44 def setUp(self):
45 self._expected_commands = []
46 self._test_temp_file = None
47 self._test_gs_base = None
48 self._test_destdir = None
49 self._test_gs_acl = None
50 self._local_tempdir = tempfile.mkdtemp()
51
52 def _MockCommand(command):
53 self.assertEquals(self._expected_commands.pop(0), ' '.join(command))
54
55 def _MockGSUtilFileCopy(filename, gs_base, subdir, gs_acl):
56 self.assertEquals(self._test_temp_file, filename)
57 self.assertEquals(self._test_gs_base, gs_base)
58 self.assertEquals(self._test_destdir, subdir)
59 self.assertEquals(self._test_gs_acl, gs_acl)
60
61 def _MockGSUtilDownloadFile(src, dst):
62 pass
63
64 self._original_bash_run_command = shell_utils.run
65 shell_utils.run = _MockCommand
66
67 self._original_gsutil_file_copy = slave_utils.GSUtilCopyFile
68 slave_utils.GSUtilCopyFile = _MockGSUtilFileCopy
69
70 self._original_gsutil_download_file = slave_utils.GSUtilDownloadFile
71 slave_utils.GSUtilDownloadFile = _MockGSUtilDownloadFile
72
73 self._original_file = __builtin__.open
74
75 def tearDown(self):
76 self.assertEquals(len(self._expected_commands), 0)
77 shell_utils.run = self._original_bash_run_command
78 slave_utils.GSUtilCopyFile = self._original_gsutil_file_copy
79 slave_utils.GSUtilDownloadFile = self._original_gsutil_download_file
80 __builtin__.open = self._original_file
81 shutil.rmtree(self._local_tempdir)
82
83 def test_delete_storage_object(self):
84 self._expected_commands = [('%s rm -R superman' % GSUTIL_LOCATION)]
85 gs_utils.delete_storage_object('superman')
86
87 def test_upload_file(self):
88 self._expected_commands = [(
89 '%s cp -a public /fake/local/src/path gs://fake/remote/dest/path' %
90 GSUTIL_LOCATION)]
91 gs_utils.upload_file(
92 local_src_path='/fake/local/src/path',
93 remote_dest_path='gs://fake/remote/dest/path',
94 gs_acl='public')
95
96 def test_upload_dir_contents_empty(self):
97 self._expected_commands = []
98 gs_utils.upload_dir_contents(
99 local_src_dir=self._local_tempdir, remote_dest_dir='remote_dest_dir',
100 gs_acl='public')
101
102 def test_upload_dir_contents_one_file(self):
103 """Upload src_dir containing one file, and no subdirs."""
104 self._test_upload_dir_contents(filenames=['file1'])
105
106 def test_upload_dir_contents_multiple_files(self):
107 """Upload src_dir containing multiple files, and no subdirs."""
108 self._test_upload_dir_contents(filenames=['file1', 'file2'])
109
110 def _test_upload_dir_contents(self, filenames):
111 """Helper function for upload_dir_contents() unittests.
112
113 Args:
114 filenames: basenames of files to create within local_src_dir
115 """
116 local_src_dir = self._local_tempdir
117 remote_dest_dir = 'remote_dest_dir'
118 for filename in filenames:
119 self._expected_commands.append('%s cp -a public %s %s' % (
120 GSUTIL_LOCATION,
121 os.path.join(local_src_dir, filename),
122 posixpath.join(remote_dest_dir, filename)))
123 with open(os.path.join(local_src_dir, filename), 'w'):
124 pass
125 gs_utils.upload_dir_contents(
126 local_src_dir=local_src_dir, remote_dest_dir=remote_dest_dir,
127 gs_acl='public')
128
129 def test_upload_dir_contents_one_dir(self):
130 """Upload src_dir containing a subdir, which in turn contains files."""
131 local_src_dir = self._local_tempdir
132 remote_dest_dir = 'remote_dest_dir'
133 subdir = 'subdir'
134 os.mkdir(os.path.join(local_src_dir, subdir))
135 for filename in ['file1', 'file2']:
136 self._expected_commands.append('%s cp -a public %s %s' % (
137 GSUTIL_LOCATION,
138 os.path.join(local_src_dir, subdir, filename),
139 posixpath.join(remote_dest_dir, subdir, filename)))
140 with open(os.path.join(local_src_dir, subdir, filename), 'w'):
141 pass
142 gs_utils.upload_dir_contents(
143 local_src_dir=local_src_dir, remote_dest_dir=remote_dest_dir,
144 gs_acl='public')
145
146 def test_download_dir_contents(self):
147 self._expected_commands = [(
148 '%s -m cp -R superman batman' % GSUTIL_LOCATION)]
149 gs_utils.download_dir_contents('superman', 'batman')
150
151 def test_copy_dir_contents(self):
152 self._expected_commands = [(
153 '%s -m cp -a public -R superman batman' % GSUTIL_LOCATION)]
154 gs_utils.copy_dir_contents('superman', 'batman', 'public')
155
156 def test_does_storage_object_exist(self):
157 self._expected_commands = [('%s ls superman' % GSUTIL_LOCATION)]
158 gs_utils.does_storage_object_exist('superman')
159
160 def test_write_timestamp_file(self):
161 self._test_temp_file = os.path.join(tempfile.gettempdir(), 'TIMESTAMP')
162 self._test_gs_base = 'gs://test'
163 self._test_destdir = 'testdir'
164 self._test_gs_acl = 'private'
165 gs_utils.write_timestamp_file(
166 timestamp_file_name='TIMESTAMP',
167 timestamp_value=time.time(),
168 gs_base=self._test_gs_base,
169 gs_relative_dir=self._test_destdir,
170 gs_acl=self._test_gs_acl,
171 local_dir=self._local_tempdir)
172
173 def test_AreTimeStampsEqual(self):
174 self._test_gs_base = 'gs://test'
175 self._test_destdir = 'testdir'
176 local_dir = self._local_tempdir
177
178 class _MockFile():
179 def __init__(self, name, attributes):
180 self._name = name
181
182 def readlines(self):
183 return []
184
185 def read(self, arg1=None):
186 if self._name == os.path.join(tempfile.gettempdir(), 'TIMESTAMP'):
187 return TEST_TIMESTAMP
188 else:
189 return TEST_TIMESTAMP_2
190
191 def close(self):
192 pass
193
194 def __enter__(self):
195 return self
196
197 def __exit__(self, *args):
198 pass
199
200 def write(self, string):
201 pass
202
203 __builtin__.open = _MockFile
204
205 # Will be false because the tmp directory will have no TIMESTAMP in it.
206 # pylint: disable=W0212
207 self.assertFalse(
208 gs_utils._are_timestamps_equal(
209 local_dir=local_dir,
210 gs_base=self._test_gs_base,
211 gs_relative_dir=self._test_destdir))
212
213 self._test_temp_file = os.path.join(local_dir, 'TIMESTAMP')
214
215 # Will be false because the timestamps are different.
216 # pylint: disable=W0212
217 self.assertFalse(
218 gs_utils._are_timestamps_equal(
219 local_dir=local_dir,
220 gs_base=self._test_gs_base,
221 gs_relative_dir=self._test_destdir))
222
223
224 if __name__ == '__main__':
225 unittest.main()
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698