| # Copyright (C) 2009 Google Inc. All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # * Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # * Redistributions in binary form must reproduce the above |
| # copyright notice, this list of conditions and the following disclaimer |
| # in the documentation and/or other materials provided with the |
| # distribution. |
| # * Neither the name of Google Inc. nor the names of its |
| # contributors may be used to endorse or promote products derived from |
| # this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
| # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
| # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
| # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
| # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
| # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
| # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
| # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
| # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
| # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| import errno |
| import hashlib |
| import os |
| import re |
| import StringIO |
| import unittest |
| |
| from blinkpy.common.system.filesystem import _remove_contents, _sanitize_filename |
| |
| |
| class MockFileSystem(object): |
| # pylint: disable=unused-argument |
| |
| sep = '/' |
| pardir = '..' |
| |
| def __init__(self, files=None, dirs=None, cwd='/'): |
| """Initializes a "mock" filesystem that can be used to replace the |
| FileSystem class in tests. |
| |
| Args: |
| files: A dictionary of filenames to file contents. A file contents |
| value of None indicates that the file does not exist. |
| """ |
| self.files = files or {} |
| self.executable_files = set() |
| self.written_files = {} |
| self.last_tmpdir = None |
| self.current_tmpno = 0 |
| self.cwd = cwd |
| self.dirs = set(dirs or []) |
| self.dirs.add(cwd) |
| for file_path in self.files: |
| directory = self.dirname(file_path) |
| while directory not in self.dirs: |
| self.dirs.add(directory) |
| directory = self.dirname(directory) |
| |
| def clear_written_files(self): |
| # This function can be used to track what is written between steps in a test. |
| self.written_files = {} |
| |
| def _raise_not_found(self, path): |
| raise IOError(errno.ENOENT, path, os.strerror(errno.ENOENT)) |
| |
| def _split(self, path): |
| # This is not quite a full implementation of os.path.split; see: |
| # http://docs.python.org/library/os.path.html#os.path.split |
| if self.sep in path: |
| return path.rsplit(self.sep, 1) |
| return ('', path) |
| |
| def make_executable(self, file_path): |
| self.executable_files.add(file_path) |
| |
| def abspath(self, path): |
| if os.path.isabs(path): |
| return self.normpath(path) |
| return self.abspath(self.join(self.cwd, path)) |
| |
| def realpath(self, path): |
| return self.abspath(path) |
| |
| def basename(self, path): |
| return self._split(path)[1] |
| |
| def expanduser(self, path): |
| if path[0] != '~': |
| return path |
| parts = path.split(self.sep, 1) |
| home_directory = self.sep + 'Users' + self.sep + 'mock' |
| if len(parts) == 1: |
| return home_directory |
| return home_directory + self.sep + parts[1] |
| |
| def path_to_module(self, module_name): |
| return ('/mock-checkout/third_party/blink/tools/' + |
| module_name.replace('.', '/') + '.py') |
| |
| def chdir(self, path): |
| path = self.normpath(path) |
| if not self.isdir(path): |
| raise OSError(errno.ENOENT, path, os.strerror(errno.ENOENT)) |
| self.cwd = path |
| |
| def copyfile(self, source, destination): |
| if not self.exists(source): |
| self._raise_not_found(source) |
| if self.isdir(source): |
| raise IOError(errno.EISDIR, source, os.strerror(errno.EISDIR)) |
| if self.isdir(destination): |
| raise IOError(errno.EISDIR, destination, os.strerror(errno.EISDIR)) |
| if not self.exists(self.dirname(destination)): |
| raise IOError(errno.ENOENT, destination, os.strerror(errno.ENOENT)) |
| |
| self.files[destination] = self.files[source] |
| self.written_files[destination] = self.files[source] |
| |
| def dirname(self, path): |
| return self._split(path)[0] |
| |
| def exists(self, path): |
| return self.isfile(path) or self.isdir(path) |
| |
| def files_under(self, path, dirs_to_skip=None, file_filter=None): |
| dirs_to_skip = dirs_to_skip or [] |
| |
| filter_all = lambda fs, dirpath, basename: True |
| |
| file_filter = file_filter or filter_all |
| files = [] |
| if self.isfile(path): |
| if (file_filter(self, self.dirname(path), self.basename(path)) |
| and self.files[path] is not None): |
| files.append(path) |
| return files |
| |
| if self.basename(path) in dirs_to_skip: |
| return [] |
| |
| if not path.endswith(self.sep): |
| path += self.sep |
| |
| dir_substrings = [self.sep + d + self.sep for d in dirs_to_skip] |
| for filename in self.files: |
| if not filename.startswith(path): |
| continue |
| |
| suffix = filename[len(path) - 1:] |
| if any(dir_substring in suffix |
| for dir_substring in dir_substrings): |
| continue |
| |
| dirpath, basename = self._split(filename) |
| if (file_filter(self, dirpath, basename) |
| and self.files[filename] is not None): |
| files.append(filename) |
| |
| return files |
| |
| def getcwd(self): |
| return self.cwd |
| |
| def glob(self, glob_string): |
| # FIXME: This handles '*', but not '?', '[', or ']'. |
| glob_string = re.escape(glob_string) |
| glob_string = glob_string.replace('\\*', '[^\\/]*') + '$' |
| glob_string = glob_string.replace('\\/', '/') |
| path_filter = lambda path: re.match(glob_string, path) |
| |
| # We could use fnmatch.fnmatch, but that might not do the right thing on Windows. |
| existing_files = [ |
| path for path, contents in self.files.items() |
| if contents is not None |
| ] |
| return filter(path_filter, existing_files) + filter( |
| path_filter, self.dirs) |
| |
| def isabs(self, path): |
| return path.startswith(self.sep) |
| |
| def isfile(self, path): |
| return path in self.files and self.files[path] is not None |
| |
| def isdir(self, path): |
| return self.normpath(path) in self.dirs |
| |
| def _slow_but_correct_join(self, *comps): |
| return re.sub(re.escape(os.path.sep), self.sep, os.path.join(*comps)) |
| |
| def join(self, *comps): |
| # This function is called a lot, so we optimize it; there are |
| # unit tests to check that we match _slow_but_correct_join(), above. |
| path = '' |
| sep = self.sep |
| for comp in comps: |
| if not comp: |
| continue |
| if comp[0] == sep: |
| path = comp |
| continue |
| if path: |
| path += sep |
| path += comp |
| if comps[-1] == '' and path: |
| path += '/' |
| path = path.replace(sep + sep, sep) |
| return path |
| |
| def listdir(self, path): |
| _, directories, files = list(self.walk(path))[0] |
| return directories + files |
| |
| def walk(self, top): |
| sep = self.sep |
| if not self.isdir(top): |
| raise OSError('%s is not a directory' % top) |
| |
| if not top.endswith(sep): |
| top += sep |
| |
| directories = [] |
| files = [] |
| for file_path in self.files: |
| if self.exists(file_path) and file_path.startswith(top): |
| remaining = file_path[len(top):] |
| if sep in remaining: |
| directory = remaining[:remaining.index(sep)] |
| if directory not in directories: |
| directories.append(directory) |
| else: |
| files.append(remaining) |
| file_system_tuples = [(top[:-1], directories, files)] |
| for directory in directories: |
| directory = top + directory |
| tuples_from_subdirs = self.walk(directory) |
| file_system_tuples += tuples_from_subdirs |
| return file_system_tuples |
| |
| def mtime(self, path): |
| if self.exists(path): |
| return 0 |
| self._raise_not_found(path) |
| |
| def mktemp(self, suffix='', prefix='tmp', dir=None, **_): # pylint: disable=redefined-builtin |
| if dir is None: |
| dir = self.sep + '__im_tmp' |
| curno = self.current_tmpno |
| self.current_tmpno += 1 |
| self.last_tmpdir = self.join(dir, '%s_%u_%s' % (prefix, curno, suffix)) |
| return self.last_tmpdir |
| |
| def mkdtemp(self, **kwargs): |
| class TemporaryDirectory(object): |
| def __init__(self, fs, **kwargs): |
| self._kwargs = kwargs |
| self._filesystem = fs |
| self._directory_path = fs.mktemp(**kwargs) # pylint: disable=protected-access |
| fs.maybe_make_directory(self._directory_path) |
| |
| def __str__(self): |
| return self._directory_path |
| |
| def __enter__(self): |
| return self._directory_path |
| |
| def __exit__(self, exception_type, exception_value, traceback): |
| # Only self-delete if necessary. |
| |
| # FIXME: Should we delete non-empty directories? |
| if self._filesystem.exists(self._directory_path): |
| self._filesystem.rmtree(self._directory_path) |
| |
| return TemporaryDirectory(fs=self, **kwargs) |
| |
| def maybe_make_directory(self, *path): |
| norm_path = self.normpath(self.join(*path)) |
| while norm_path and not self.isdir(norm_path): |
| self.dirs.add(norm_path) |
| norm_path = self.dirname(norm_path) |
| |
| def move(self, source, destination): |
| if not self.exists(source): |
| self._raise_not_found(source) |
| if self.isfile(source): |
| self.files[destination] = self.files[source] |
| self.written_files[destination] = self.files[destination] |
| self.files[source] = None |
| self.written_files[source] = None |
| return |
| self.copytree(source, destination) |
| self.rmtree(source) |
| |
| def _slow_but_correct_normpath(self, path): |
| return re.sub(re.escape(os.path.sep), self.sep, os.path.normpath(path)) |
| |
| def normpath(self, path): |
| # This function is called a lot, so we try to optimize the common cases |
| # instead of always calling _slow_but_correct_normpath(), above. |
| if '..' in path or '/./' in path: |
| # This doesn't happen very often; don't bother trying to optimize it. |
| return self._slow_but_correct_normpath(path) |
| if not path: |
| return '.' |
| if path == '/': |
| return path |
| if path == '/.': |
| return '/' |
| if path.endswith('/.'): |
| return path[:-2] |
| if path.endswith('/'): |
| return path[:-1] |
| return path |
| |
| def open_binary_tempfile(self, suffix=''): |
| path = self.mktemp(suffix) |
| return (WritableBinaryFileObject(self, path), path) |
| |
| def open_binary_file_for_reading(self, path): |
| if self.files[path] is None: |
| self._raise_not_found(path) |
| return ReadableBinaryFileObject(self, path, self.files[path]) |
| |
| def open_binary_file_for_writing(self, path): |
| return WritableBinaryFileObject(self, path) |
| |
| def read_binary_file(self, path): |
| # Intentionally raises KeyError if we don't recognize the path. |
| if self.files[path] is None: |
| self._raise_not_found(path) |
| return self.files[path] |
| |
| def write_binary_file(self, path, contents): |
| # FIXME: should this assert if dirname(path) doesn't exist? |
| self.maybe_make_directory(self.dirname(path)) |
| self.files[path] = contents |
| self.written_files[path] = contents |
| |
| def open_text_tempfile(self, suffix=''): |
| path = self.mktemp(suffix) |
| return (WritableTextFileObject(self, path), path) |
| |
| def open_text_file_for_reading(self, path): |
| if self.files[path] is None: |
| self._raise_not_found(path) |
| return ReadableTextFileObject(self, path, self.files[path]) |
| |
| def open_text_file_for_writing(self, path): |
| return WritableTextFileObject(self, path) |
| |
| def read_text_file(self, path): |
| return self.read_binary_file(path).decode('utf-8') |
| |
| def write_text_file(self, path, contents): |
| return self.write_binary_file(path, contents.encode('utf-8')) |
| |
| def sha1(self, path): |
| contents = self.read_binary_file(path) |
| return hashlib.sha1(contents).hexdigest() |
| |
| def relpath(self, path, start='.'): |
| # Since os.path.relpath() calls os.path.normpath() |
| # (see http://docs.python.org/library/os.path.html#os.path.abspath ) |
| # it also removes trailing slashes and converts forward and backward |
| # slashes to the preferred slash os.sep. |
| start = self.abspath(start) |
| path = self.abspath(path) |
| |
| common_root = start |
| dot_dot = '' |
| while not common_root == '': |
| if path.startswith(common_root): |
| break |
| common_root = self.dirname(common_root) |
| dot_dot += '..' + self.sep |
| |
| rel_path = path[len(common_root):] |
| |
| if not rel_path: |
| return '.' |
| |
| if rel_path[0] == self.sep: |
| # It is probably sufficient to remove just the first character |
| # since os.path.normpath() collapses separators, but we use |
| # lstrip() just to be sure. |
| rel_path = rel_path.lstrip(self.sep) |
| elif not common_root == '/': |
| # We are in the case typified by the following example: |
| # path = "/tmp/foobar", start = "/tmp/foo" -> rel_path = "bar" |
| common_root = self.dirname(common_root) |
| dot_dot += '..' + self.sep |
| rel_path = path[len(common_root) + 1:] |
| |
| return dot_dot + rel_path |
| |
| def remove(self, path, retry=True): |
| if self.files[path] is None: |
| self._raise_not_found(path) |
| self.files[path] = None |
| self.written_files[path] = None |
| |
| def rmtree(self, path_to_remove, ignore_errors=True, onerror=None): |
| path_to_remove = self.normpath(path_to_remove) |
| |
| for file_path in self.files: |
| # We need to add a trailing separator to path_to_remove to avoid matching |
| # cases like path_to_remove='/foo/b' and file_path='/foo/bar/baz'. |
| if file_path == path_to_remove or file_path.startswith( |
| path_to_remove + self.sep): |
| self.files[file_path] = None |
| |
| def should_remove(directory): |
| return directory == path_to_remove or directory.startswith( |
| path_to_remove + self.sep) |
| |
| self.dirs = {d for d in self.dirs if not should_remove(d)} |
| |
| def remove_contents(self, dirname): |
| return _remove_contents(self, dirname, sleep=lambda *args, **kw: None) |
| |
| def copytree(self, source, destination): |
| source = self.normpath(source) |
| destination = self.normpath(destination) |
| |
| for source_file in list(self.files): |
| if source_file.startswith(source): |
| destination_path = self.join(destination, |
| self.relpath(source_file, source)) |
| self.maybe_make_directory(self.dirname(destination_path)) |
| self.files[destination_path] = self.files[source_file] |
| |
| def split(self, path): |
| idx = path.rfind(self.sep) |
| if idx == -1: |
| return ('', path) |
| return (path[:idx], path[(idx + 1):]) |
| |
| def splitext(self, path): |
| idx = path.rfind('.') |
| if idx == -1: |
| idx = len(path) |
| return (path[0:idx], path[idx:]) |
| |
| def symlink(self, source, link_name): |
| raise NotImplementedError('Symlink not expected to be called in tests') |
| |
| def sanitize_filename(self, filename, replacement='_'): |
| return _sanitize_filename(filename, replacement) |
| |
| |
| class WritableBinaryFileObject(object): |
| def __init__(self, fs, path): |
| self.fs = fs |
| self.path = path |
| self.closed = False |
| self.fs.files[path] = '' |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, exception_type, exception_value, traceback): |
| self.close() |
| |
| def close(self): |
| self.closed = True |
| |
| def write(self, string): |
| self.fs.files[self.path] += string |
| self.fs.written_files[self.path] = self.fs.files[self.path] |
| |
| |
| class WritableTextFileObject(WritableBinaryFileObject): |
| def write(self, string): |
| WritableBinaryFileObject.write(self, string.encode('utf-8')) |
| |
| def writelines(self, lines): |
| self.fs.files[self.path] = "".join(lines).encode('utf-8') |
| self.fs.written_files[self.path] = self.fs.files[self.path] |
| |
| |
| class ReadableBinaryFileObject(object): |
| def __init__(self, fs, path, data): |
| self.fs = fs |
| self.path = path |
| self.closed = False |
| self.data = data |
| self.offset = 0 |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, exception_type, exception_value, traceback): |
| self.close() |
| |
| def close(self): |
| self.closed = True |
| |
| def read(self, num_bytes=None): |
| if not num_bytes: |
| return self.data[self.offset:] |
| start = self.offset |
| self.offset += num_bytes |
| return self.data[start:self.offset] |
| |
| def seek(self, offset, whence=os.SEEK_SET): |
| if whence == os.SEEK_SET: |
| self.offset = offset |
| elif whence == os.SEEK_CUR: |
| self.offset += offset |
| elif whence == os.SEEK_END: |
| self.offset = len(self.data) + offset |
| else: |
| assert False, "Unknown seek mode %s" % whence |
| |
| |
| class ReadableTextFileObject(ReadableBinaryFileObject): |
| def __init__(self, fs, path, data): |
| super(ReadableTextFileObject, self).__init__( |
| fs, path, StringIO.StringIO(data.decode('utf-8'))) |
| |
| def close(self): |
| self.data.close() |
| super(ReadableTextFileObject, self).close() |
| |
| def read(self, num_bytes=-1): |
| return self.data.read(num_bytes) |
| |
| def readline(self, length=None): |
| return self.data.readline(length) |
| |
| def readlines(self): |
| return self.data.readlines() |
| |
| def __iter__(self): |
| return self.data.__iter__() |
| |
| def next(self): |
| return self.data.next() |
| |
| def seek(self, offset, whence=os.SEEK_SET): |
| self.data.seek(offset, whence) |
| |
| |
| class FileSystemTestCase(unittest.TestCase): |
| # pylint: disable=invalid-name |
| # Use assertFilesAdded to be consistent with unittest. |
| |
| class _AssertFilesAddedContext(object): |
| """Internal class used by FileTestCase.assertFilesAdded().""" |
| |
| def __init__(self, test_case, mock_filesystem, expected_files): |
| self.test_case = test_case |
| self.mock_filesystem = mock_filesystem |
| self.expected_files = expected_files |
| |
| def __enter__(self): |
| # Make sure that the expected_files aren't already in the mock |
| # file system. |
| for filepath in self.expected_files: |
| assert filepath not in self.mock_filesystem.files, "%s was already in mock file system (%r)" % ( |
| filepath, self.mock_filesystem.files) |
| return self |
| |
| def __exit__(self, exc_type, exc_value, tb): |
| # Exception already occurring, just exit. |
| if exc_type is not None: |
| return |
| |
| for filepath in sorted(self.expected_files): |
| self.test_case.assertIn(filepath, self.mock_filesystem.files) |
| self.test_case.assertEqual( |
| self.expected_files[filepath], |
| self.mock_filesystem.files[filepath]) |
| |
| def assertFilesAdded(self, mock_filesystem, files): |
| """Assert that the given files where added to the mock_filesystem. |
| |
| Use in a similar manner to self.assertRaises; |
| |
| with self.assertFilesAdded(mock_filesystem, {'/newfile': 'contents'}): |
| code(mock_filesystem) |
| """ |
| return self._AssertFilesAddedContext(self, mock_filesystem, files) |
