blob: 4bb22b60af27922472d0ea6c700cd9d1208130f9 [file] [log] [blame]
maruel@chromium.org561d4b22013-09-26 21:08:08 +00001#!/usr/bin/env python
2# coding=utf-8
maruelea586f32016-04-05 11:11:33 -07003# Copyright 2013 The LUCI Authors. All rights reserved.
maruelf1f5e2a2016-05-25 17:10:39 -07004# Use of this source code is governed under the Apache License, Version 2.0
5# that can be found in the LICENSE file.
maruel@chromium.org561d4b22013-09-26 21:08:08 +00006
maruel9cdd7612015-12-02 13:40:52 -08007import getpass
maruel@chromium.org561d4b22013-09-26 21:08:08 +00008import logging
9import os
10import tempfile
11import unittest
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040012import StringIO
13import subprocess
maruel@chromium.org561d4b22013-09-26 21:08:08 +000014import sys
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040015import time
maruel@chromium.org561d4b22013-09-26 21:08:08 +000016
Marc-Antoine Ruel0eb2eb22019-01-29 21:00:16 +000017FILE_PATH = os.path.abspath(__file__.decode(sys.getfilesystemencoding()))
18BASE_DIR = os.path.dirname(FILE_PATH)
maruel@chromium.org561d4b22013-09-26 21:08:08 +000019ROOT_DIR = os.path.dirname(BASE_DIR)
20sys.path.insert(0, ROOT_DIR)
Marc-Antoine Ruelf1d827c2014-11-24 15:22:25 -050021sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party'))
maruel@chromium.org561d4b22013-09-26 21:08:08 +000022
Marc-Antoine Ruelf1d827c2014-11-24 15:22:25 -050023from depot_tools import auto_stub
maruel4b14f042015-10-06 12:08:08 -070024from depot_tools import fix_encoding
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040025import test_utils
maruel@chromium.org561d4b22013-09-26 21:08:08 +000026from utils import file_path
maruel4e732992015-10-16 10:17:21 -070027from utils import fs
maruel@chromium.org561d4b22013-09-26 21:08:08 +000028
29
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040030def write_content(filepath, content):
maruel4e732992015-10-16 10:17:21 -070031 with fs.open(filepath, 'wb') as f:
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040032 f.write(content)
33
34
Marc-Antoine Ruelf1d827c2014-11-24 15:22:25 -050035class FilePathTest(auto_stub.TestCase):
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040036 def setUp(self):
37 super(FilePathTest, self).setUp()
38 self._tempdir = None
39
40 def tearDown(self):
maruel4b14f042015-10-06 12:08:08 -070041 try:
42 if self._tempdir:
maruel4e732992015-10-16 10:17:21 -070043 for dirpath, dirnames, filenames in fs.walk(
maruel4b14f042015-10-06 12:08:08 -070044 self._tempdir, topdown=True):
45 for filename in filenames:
46 file_path.set_read_only(os.path.join(dirpath, filename), False)
47 for dirname in dirnames:
48 file_path.set_read_only(os.path.join(dirpath, dirname), False)
49 file_path.rmtree(self._tempdir)
50 finally:
51 super(FilePathTest, self).tearDown()
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040052
53 @property
54 def tempdir(self):
55 if not self._tempdir:
Marc-Antoine Ruel0eb2eb22019-01-29 21:00:16 +000056 self._tempdir = tempfile.mkdtemp(prefix=u'file_path_test')
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040057 return self._tempdir
58
vadimshe42aeba2016-06-03 12:32:21 -070059 def test_atomic_replace_new_file(self):
60 path = os.path.join(self.tempdir, 'new_file')
61 file_path.atomic_replace(path, 'blah')
62 with open(path, 'rb') as f:
63 self.assertEqual('blah', f.read())
64 self.assertEqual([u'new_file'], os.listdir(self.tempdir))
65
66 def test_atomic_replace_existing_file(self):
67 path = os.path.join(self.tempdir, 'existing_file')
68 with open(path, 'wb') as f:
69 f.write('existing body')
70 file_path.atomic_replace(path, 'new body')
71 with open(path, 'rb') as f:
72 self.assertEqual('new body', f.read())
73 self.assertEqual([u'existing_file'], os.listdir(self.tempdir))
74
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040075 def assertFileMode(self, filepath, mode, umask=None):
76 umask = test_utils.umask() if umask is None else umask
maruel4e732992015-10-16 10:17:21 -070077 actual = fs.stat(filepath).st_mode
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -040078 expected = mode & ~umask
79 self.assertEqual(
80 expected,
81 actual,
82 (filepath, oct(expected), oct(actual), oct(umask)))
83
84 def assertMaskedFileMode(self, filepath, mode):
85 """It's usually when the file was first marked read only."""
86 self.assertFileMode(filepath, mode, 0 if sys.platform == 'win32' else 077)
87
maruel@chromium.org561d4b22013-09-26 21:08:08 +000088 def test_native_case_end_with_os_path_sep(self):
89 # Make sure the trailing os.path.sep is kept.
90 path = file_path.get_native_path_case(ROOT_DIR) + os.path.sep
91 self.assertEqual(file_path.get_native_path_case(path), path)
92
93 def test_native_case_end_with_dot_os_path_sep(self):
94 path = file_path.get_native_path_case(ROOT_DIR + os.path.sep)
95 self.assertEqual(
96 file_path.get_native_path_case(path + '.' + os.path.sep),
97 path)
98
99 def test_native_case_non_existing(self):
100 # Make sure it doesn't throw on non-existing files.
101 non_existing = 'trace_input_test_this_file_should_not_exist'
102 path = os.path.expanduser('~/' + non_existing)
103 self.assertFalse(os.path.exists(path))
104 path = file_path.get_native_path_case(ROOT_DIR) + os.path.sep
105 self.assertEqual(file_path.get_native_path_case(path), path)
106
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400107 def test_delete_wd_rf(self):
108 # Confirms that a RO file in a RW directory can be deleted on non-Windows.
109 dir_foo = os.path.join(self.tempdir, 'foo')
110 file_bar = os.path.join(dir_foo, 'bar')
maruel4e732992015-10-16 10:17:21 -0700111 fs.mkdir(dir_foo, 0777)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400112 write_content(file_bar, 'bar')
113 file_path.set_read_only(dir_foo, False)
114 file_path.set_read_only(file_bar, True)
115 self.assertFileMode(dir_foo, 040777)
116 self.assertMaskedFileMode(file_bar, 0100444)
117 if sys.platform == 'win32':
118 # On Windows, a read-only file can't be deleted.
119 with self.assertRaises(OSError):
maruel4e732992015-10-16 10:17:21 -0700120 fs.remove(file_bar)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400121 else:
maruel4e732992015-10-16 10:17:21 -0700122 fs.remove(file_bar)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400123
124 def test_delete_rd_wf(self):
125 # Confirms that a Rw file in a RO directory can be deleted on Windows only.
126 dir_foo = os.path.join(self.tempdir, 'foo')
127 file_bar = os.path.join(dir_foo, 'bar')
maruel4e732992015-10-16 10:17:21 -0700128 fs.mkdir(dir_foo, 0777)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400129 write_content(file_bar, 'bar')
130 file_path.set_read_only(dir_foo, True)
131 file_path.set_read_only(file_bar, False)
132 self.assertMaskedFileMode(dir_foo, 040555)
133 self.assertFileMode(file_bar, 0100666)
134 if sys.platform == 'win32':
135 # A read-only directory has a convoluted meaning on Windows, it means that
136 # the directory is "personalized". This is used as a signal by Windows
137 # Explorer to tell it to look into the directory for desktop.ini.
138 # See http://support.microsoft.com/kb/326549 for more details.
139 # As such, it is important to not try to set the read-only bit on
140 # directories on Windows since it has no effect other than trigger
141 # Windows Explorer to look for desktop.ini, which is unnecessary.
maruel4e732992015-10-16 10:17:21 -0700142 fs.remove(file_bar)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400143 else:
144 with self.assertRaises(OSError):
maruel4e732992015-10-16 10:17:21 -0700145 fs.remove(file_bar)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400146
147 def test_delete_rd_rf(self):
148 # Confirms that a RO file in a RO directory can't be deleted.
149 dir_foo = os.path.join(self.tempdir, 'foo')
150 file_bar = os.path.join(dir_foo, 'bar')
maruel4e732992015-10-16 10:17:21 -0700151 fs.mkdir(dir_foo, 0777)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400152 write_content(file_bar, 'bar')
153 file_path.set_read_only(dir_foo, True)
154 file_path.set_read_only(file_bar, True)
155 self.assertMaskedFileMode(dir_foo, 040555)
156 self.assertMaskedFileMode(file_bar, 0100444)
157 with self.assertRaises(OSError):
158 # It fails for different reason depending on the OS. See the test cases
159 # above.
maruel4e732992015-10-16 10:17:21 -0700160 fs.remove(file_bar)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400161
162 def test_hard_link_mode(self):
163 # Creates a hard link, see if the file mode changed on the node or the
164 # directory entry.
165 dir_foo = os.path.join(self.tempdir, 'foo')
166 file_bar = os.path.join(dir_foo, 'bar')
167 file_link = os.path.join(dir_foo, 'link')
maruel4e732992015-10-16 10:17:21 -0700168 fs.mkdir(dir_foo, 0777)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400169 write_content(file_bar, 'bar')
170 file_path.hardlink(file_bar, file_link)
171 self.assertFileMode(file_bar, 0100666)
172 self.assertFileMode(file_link, 0100666)
173 file_path.set_read_only(file_bar, True)
174 self.assertMaskedFileMode(file_bar, 0100444)
175 self.assertMaskedFileMode(file_link, 0100444)
176 # This is bad news for Windows; on Windows, the file must be writeable to be
177 # deleted, but the file node is modified. This means that every hard links
178 # must be reset to be read-only after deleting one of the hard link
179 # directory entry.
180
Marc-Antoine Ruel0a795bd2015-01-16 20:32:10 -0500181 def test_rmtree_unicode(self):
182 subdir = os.path.join(self.tempdir, 'hi')
maruel4e732992015-10-16 10:17:21 -0700183 fs.mkdir(subdir)
Marc-Antoine Ruel0a795bd2015-01-16 20:32:10 -0500184 filepath = os.path.join(
185 subdir, u'\u0627\u0644\u0635\u064A\u0646\u064A\u0629')
maruel4e732992015-10-16 10:17:21 -0700186 with fs.open(filepath, 'wb') as f:
Marc-Antoine Ruel0a795bd2015-01-16 20:32:10 -0500187 f.write('hi')
188 # In particular, it fails when the input argument is a str.
189 file_path.rmtree(str(subdir))
190
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400191 if sys.platform == 'darwin':
192 def test_native_case_symlink_wrong_case(self):
193 base_dir = file_path.get_native_path_case(BASE_DIR)
194 trace_inputs_dir = os.path.join(base_dir, 'trace_inputs')
195 actual = file_path.get_native_path_case(trace_inputs_dir)
196 self.assertEqual(trace_inputs_dir, actual)
197
198 # Make sure the symlink is not resolved.
199 data = os.path.join(trace_inputs_dir, 'Files2')
200 actual = file_path.get_native_path_case(data)
201 self.assertEqual(
202 os.path.join(trace_inputs_dir, 'files2'), actual)
203
204 data = os.path.join(trace_inputs_dir, 'Files2', '')
205 actual = file_path.get_native_path_case(data)
206 self.assertEqual(
207 os.path.join(trace_inputs_dir, 'files2', ''), actual)
208
209 data = os.path.join(trace_inputs_dir, 'Files2', 'Child1.py')
210 actual = file_path.get_native_path_case(data)
211 # TODO(maruel): Should be child1.py.
212 self.assertEqual(
213 os.path.join(trace_inputs_dir, 'files2', 'Child1.py'), actual)
214
maruel@chromium.org561d4b22013-09-26 21:08:08 +0000215 if sys.platform in ('darwin', 'win32'):
216 def test_native_case_not_sensitive(self):
217 # The home directory is almost guaranteed to have mixed upper/lower case
218 # letters on both Windows and OSX.
219 # This test also ensures that the output is independent on the input
220 # string case.
221 path = os.path.expanduser(u'~')
222 self.assertTrue(os.path.isdir(path))
223 path = path.replace('/', os.path.sep)
224 if sys.platform == 'win32':
225 # Make sure the drive letter is upper case for consistency.
226 path = path[0].upper() + path[1:]
227 # This test assumes the variable is in the native path case on disk, this
228 # should be the case. Verify this assumption:
229 self.assertEqual(path, file_path.get_native_path_case(path))
230 self.assertEqual(
231 file_path.get_native_path_case(path.lower()),
232 file_path.get_native_path_case(path.upper()))
233
234 def test_native_case_not_sensitive_non_existent(self):
235 # This test also ensures that the output is independent on the input
236 # string case.
237 non_existing = os.path.join(
238 'trace_input_test_this_dir_should_not_exist', 'really not', '')
239 path = os.path.expanduser(os.path.join(u'~', non_existing))
240 path = path.replace('/', os.path.sep)
maruel4e732992015-10-16 10:17:21 -0700241 self.assertFalse(fs.exists(path))
maruel@chromium.org561d4b22013-09-26 21:08:08 +0000242 lower = file_path.get_native_path_case(path.lower())
243 upper = file_path.get_native_path_case(path.upper())
244 # Make sure non-existing element is not modified:
245 self.assertTrue(lower.endswith(non_existing.lower()))
246 self.assertTrue(upper.endswith(non_existing.upper()))
247 self.assertEqual(lower[:-len(non_existing)], upper[:-len(non_existing)])
248
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400249 if sys.platform == 'win32':
250 def test_native_case_alternate_datastream(self):
251 # Create the file manually, since tempfile doesn't support ADS.
Marc-Antoine Ruel3c979cb2015-03-11 13:43:28 -0400252 tempdir = unicode(tempfile.mkdtemp(prefix=u'trace_inputs'))
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400253 try:
254 tempdir = file_path.get_native_path_case(tempdir)
255 basename = 'foo.txt'
256 filename = basename + ':Zone.Identifier'
257 filepath = os.path.join(tempdir, filename)
258 open(filepath, 'w').close()
259 self.assertEqual(filepath, file_path.get_native_path_case(filepath))
260 data_suffix = ':$DATA'
261 self.assertEqual(
262 filepath + data_suffix,
263 file_path.get_native_path_case(filepath + data_suffix))
264
265 open(filepath + '$DATA', 'w').close()
266 self.assertEqual(
267 filepath + data_suffix,
268 file_path.get_native_path_case(filepath + data_suffix))
269 # Ensure the ADS weren't created as separate file. You love NTFS, don't
270 # you?
maruel4e732992015-10-16 10:17:21 -0700271 self.assertEqual([basename], fs.listdir(tempdir))
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400272 finally:
maruel4b14f042015-10-06 12:08:08 -0700273 file_path.rmtree(tempdir)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400274
275 def test_rmtree_win(self):
276 # Mock our sleep for faster test case execution.
277 sleeps = []
278 self.mock(time, 'sleep', sleeps.append)
279 self.mock(sys, 'stderr', StringIO.StringIO())
280
281 # Open a child process, so the file is locked.
282 subdir = os.path.join(self.tempdir, 'to_be_deleted')
maruel4e732992015-10-16 10:17:21 -0700283 fs.mkdir(subdir)
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400284 script = 'import time; open(\'a\', \'w\'); time.sleep(60)'
285 proc = subprocess.Popen([sys.executable, '-c', script], cwd=subdir)
286 try:
287 # Wait until the file exist.
maruel4e732992015-10-16 10:17:21 -0700288 while not fs.isfile(os.path.join(subdir, 'a')):
Marc-Antoine Ruele4ad07e2014-10-15 20:22:29 -0400289 self.assertEqual(None, proc.poll())
290 file_path.rmtree(subdir)
291 self.assertEqual([2, 4, 2], sleeps)
292 # sys.stderr.getvalue() would return a fair amount of output but it is
293 # not completely deterministic so we're not testing it here.
294 finally:
295 proc.wait()
296
Marc-Antoine Ruela275b292014-11-25 15:17:21 -0500297 def test_filter_processes_dir_win(self):
298 python_dir = os.path.dirname(sys.executable)
Marc-Antoine Ruel0eb2eb22019-01-29 21:00:16 +0000299 processes = file_path._filter_processes_dir_win(
300 file_path._enum_processes_win(), python_dir)
Marc-Antoine Ruela275b292014-11-25 15:17:21 -0500301 self.assertTrue(processes)
302 proc_names = [proc.ExecutablePath for proc in processes]
303 # Try to find at least one python process.
304 self.assertTrue(
305 any(proc == sys.executable for proc in proc_names), proc_names)
306
307 def test_filter_processes_tree_win(self):
308 # Create a grand-child.
309 script = (
310 'import subprocess,sys;'
311 'proc = subprocess.Popen('
312 '[sys.executable, \'-u\', \'-c\', \'import time; print(1); '
313 'time.sleep(60)\'], stdout=subprocess.PIPE); '
314 # Signal grand child is ready.
315 'print(proc.stdout.read(1)); '
316 # Wait for parent to have completed the test.
317 'sys.stdin.read(1); '
318 'proc.kill()'
319 )
320 proc = subprocess.Popen(
321 [sys.executable, '-u', '-c', script],
322 stdin=subprocess.PIPE,
323 stdout=subprocess.PIPE)
324 try:
325 proc.stdout.read(1)
326 processes = file_path.filter_processes_tree_win(
Marc-Antoine Ruel0eb2eb22019-01-29 21:00:16 +0000327 file_path._enum_processes_win())
Marc-Antoine Ruela275b292014-11-25 15:17:21 -0500328 self.assertEqual(3, len(processes), processes)
329 proc.stdin.write('a')
330 proc.wait()
331 except Exception:
332 proc.kill()
333 finally:
334 proc.wait()
335
maruel@chromium.org561d4b22013-09-26 21:08:08 +0000336 if sys.platform != 'win32':
337 def test_symlink(self):
338 # This test will fail if the checkout is in a symlink.
339 actual = file_path.split_at_symlink(None, ROOT_DIR)
340 expected = (ROOT_DIR, None, None)
341 self.assertEqual(expected, actual)
342
343 actual = file_path.split_at_symlink(
344 None, os.path.join(BASE_DIR, 'trace_inputs'))
345 expected = (
346 os.path.join(BASE_DIR, 'trace_inputs'), None, None)
347 self.assertEqual(expected, actual)
348
349 actual = file_path.split_at_symlink(
350 None, os.path.join(BASE_DIR, 'trace_inputs', 'files2'))
351 expected = (
352 os.path.join(BASE_DIR, 'trace_inputs'), 'files2', '')
353 self.assertEqual(expected, actual)
354
355 actual = file_path.split_at_symlink(
356 ROOT_DIR, os.path.join('tests', 'trace_inputs', 'files2'))
357 expected = (
358 os.path.join('tests', 'trace_inputs'), 'files2', '')
359 self.assertEqual(expected, actual)
360 actual = file_path.split_at_symlink(
361 ROOT_DIR, os.path.join('tests', 'trace_inputs', 'files2', 'bar'))
362 expected = (
363 os.path.join('tests', 'trace_inputs'), 'files2', '/bar')
364 self.assertEqual(expected, actual)
365
366 def test_native_case_symlink_right_case(self):
367 actual = file_path.get_native_path_case(
368 os.path.join(BASE_DIR, 'trace_inputs'))
369 self.assertEqual('trace_inputs', os.path.basename(actual))
370
371 # Make sure the symlink is not resolved.
372 actual = file_path.get_native_path_case(
373 os.path.join(BASE_DIR, 'trace_inputs', 'files2'))
374 self.assertEqual('files2', os.path.basename(actual))
375
maruel9cdd7612015-12-02 13:40:52 -0800376 else:
377 def test_undeleteable_chmod(self):
378 # Create a file and a directory with an empty ACL. Then try to delete it.
379 dirpath = os.path.join(self.tempdir, 'd')
380 filepath = os.path.join(dirpath, 'f')
381 os.mkdir(dirpath)
382 with open(filepath, 'w') as f:
383 f.write('hi')
384 os.chmod(filepath, 0)
385 os.chmod(dirpath, 0)
386 file_path.rmtree(dirpath)
387
388 def test_undeleteable_owner(self):
389 # Create a file and a directory with an empty ACL. Then try to delete it.
390 dirpath = os.path.join(self.tempdir, 'd')
391 filepath = os.path.join(dirpath, 'f')
392 os.mkdir(dirpath)
393 with open(filepath, 'w') as f:
394 f.write('hi')
395 import win32security
396 user, _domain, _type = win32security.LookupAccountName(
397 '', getpass.getuser())
398 sd = win32security.SECURITY_DESCRIPTOR()
399 sd.Initialize()
400 sd.SetSecurityDescriptorOwner(user, False)
401 # Create an empty DACL, which removes all rights.
402 dacl = win32security.ACL()
403 dacl.Initialize()
404 sd.SetSecurityDescriptorDacl(1, dacl, 0)
405 win32security.SetFileSecurity(
406 fs.extend(filepath), win32security.DACL_SECURITY_INFORMATION, sd)
407 win32security.SetFileSecurity(
408 fs.extend(dirpath), win32security.DACL_SECURITY_INFORMATION, sd)
409 file_path.rmtree(dirpath)
410
maruel@chromium.org561d4b22013-09-26 21:08:08 +0000411
412if __name__ == '__main__':
maruel4b14f042015-10-06 12:08:08 -0700413 fix_encoding.fix_encoding()
maruel@chromium.org561d4b22013-09-26 21:08:08 +0000414 logging.basicConfig(
415 level=logging.DEBUG if '-v' in sys.argv else logging.ERROR)
416 if '-v' in sys.argv:
417 unittest.TestCase.maxDiff = None
418 unittest.main()