src/third_party/WebKit/Tools/Scripts/webkitpy/common/system/filesystem_mock.py

   1 # Copyright (C) 2009 Google Inc. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or without
   4 # modification, are permitted provided that the following conditions are
   5 # met:
   6 #
   7 #    * Redistributions of source code must retain the above copyright
   8 # notice, this list of conditions and the following disclaimer.
   9 #    * Redistributions in binary form must reproduce the above
  10 # copyright notice, this list of conditions and the following disclaimer
  11 # in the documentation and/or other materials provided with the
  12 # distribution.
  13 #    * Neither the name of Google Inc. nor the names of its
  14 # contributors may be used to endorse or promote products derived from
  15 # this software without specific prior written permission.
  16 #
  17 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  18 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  19 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  20 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  21 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  24 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  25 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  26 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  27 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28
  29 import StringIO
  30 import errno
  31 import hashlib
  32 import os
  33 import re
  34
  35 from webkitpy.common.system import path
  36
  37
  38 class MockFileSystem(object):
  39     sep = '/'
  40     pardir = '..'
  41
  42     def __init__(self, files=None, dirs=None, cwd='/'):
  43         """Initializes a "mock" filesystem that can be used to completely
  44         stub out a filesystem.
  45
  46         Args:
  47             files: a dict of filenames -> file contents. A file contents
  48                 value of None is used to indicate that the file should
  49                 not exist.
  50         """
  51         self.files = files or {}
  52         self.written_files = {}
  53         self.last_tmpdir = None
  54         self.current_tmpno = 0
  55         self.cwd = cwd
  56         self.dirs = set(dirs or [])
  57         self.dirs.add(cwd)
  58         for f in self.files:
  59             d = self.dirname(f)
  60             while not d in self.dirs:
  61                 self.dirs.add(d)
  62                 d = self.dirname(d)
  63
  64     def clear_written_files(self):
  65         # This function can be used to track what is written between steps in a test.
  66         self.written_files = {}
  67
  68     def _raise_not_found(self, path):
  69         raise IOError(errno.ENOENT, path, os.strerror(errno.ENOENT))
  70
  71     def _split(self, path):
  72         # This is not quite a full implementation of os.path.split
  73         # http://docs.python.org/library/os.path.html#os.path.split
  74         if self.sep in path:
  75             return path.rsplit(self.sep, 1)
  76         return ('', path)
  77
  78     def abspath(self, path):
  79         if os.path.isabs(path):
  80             return self.normpath(path)
  81         return self.abspath(self.join(self.cwd, path))
  82
  83     def realpath(self, path):
  84         return self.abspath(path)
  85
  86     def basename(self, path):
  87         return self._split(path)[1]
  88
  89     def expanduser(self, path):
  90         if path[0] != "~":
  91             return path
  92         parts = path.split(self.sep, 1)
  93         home_directory = self.sep + "Users" + self.sep + "mock"
  94         if len(parts) == 1:
  95             return home_directory
  96         return home_directory + self.sep + parts[1]
  97
  98     def path_to_module(self, module_name):
  99         return "/mock-checkout/third_party/WebKit/Tools/Scripts/" + module_name.replace('.', '/') + ".py"
 100
 101     def chdir(self, path):
 102         path = self.normpath(path)
 103         if not self.isdir(path):
 104             raise OSError(errno.ENOENT, path, os.strerror(errno.ENOENT))
 105         self.cwd = path
 106
 107     def copyfile(self, source, destination):
 108         if not self.exists(source):
 109             self._raise_not_found(source)
 110         if self.isdir(source):
 111             raise IOError(errno.EISDIR, source, os.strerror(errno.EISDIR))
 112         if self.isdir(destination):
 113             raise IOError(errno.EISDIR, destination, os.strerror(errno.EISDIR))
 114         if not self.exists(self.dirname(destination)):
 115             raise IOError(errno.ENOENT, destination, os.strerror(errno.ENOENT))
 116
 117         self.files[destination] = self.files[source]
 118         self.written_files[destination] = self.files[source]
 119
 120     def dirname(self, path):
 121         return self._split(path)[0]
 122
 123     def exists(self, path):
 124         return self.isfile(path) or self.isdir(path)
 125
 126     def files_under(self, path, dirs_to_skip=[], file_filter=None):
 127         def filter_all(fs, dirpath, basename):
 128             return True
 129
 130         file_filter = file_filter or filter_all
 131         files = []
 132         if self.isfile(path):
 133             if file_filter(self, self.dirname(path), self.basename(path)) and self.files[path] is not None:
 134                 files.append(path)
 135             return files
 136
 137         if self.basename(path) in dirs_to_skip:
 138             return []
 139
 140         if not path.endswith(self.sep):
 141             path += self.sep
 142
 143         dir_substrings = [self.sep + d + self.sep for d in dirs_to_skip]
 144         for filename in self.files:
 145             if not filename.startswith(path):
 146                 continue
 147
 148             suffix = filename[len(path) - 1:]
 149             if any(dir_substring in suffix for dir_substring in dir_substrings):
 150                 continue
 151
 152             dirpath, basename = self._split(filename)
 153             if file_filter(self, dirpath, basename) and self.files[filename] is not None:
 154                 files.append(filename)
 155
 156         return files
 157
 158     def getcwd(self):
 159         return self.cwd
 160
 161     def glob(self, glob_string):
 162         # FIXME: This handles '*', but not '?', '[', or ']'.
 163         glob_string = re.escape(glob_string)
 164         glob_string = glob_string.replace('\\*', '[^\\/]*') + '$'
 165         glob_string = glob_string.replace('\\/', '/')
 166         path_filter = lambda path: re.match(glob_string, path)
 167
 168         # We could use fnmatch.fnmatch, but that might not do the right thing on windows.
 169         existing_files = [path for path, contents in self.files.items() if contents is not None]
 170         return filter(path_filter, existing_files) + filter(path_filter, self.dirs)
 171
 172     def isabs(self, path):
 173         return path.startswith(self.sep)
 174
 175     def isfile(self, path):
 176         return path in self.files and self.files[path] is not None
 177
 178     def isdir(self, path):
 179         return self.normpath(path) in self.dirs
 180
 181     def _slow_but_correct_join(self, *comps):
 182         return re.sub(re.escape(os.path.sep), self.sep, os.path.join(*comps))
 183
 184     def join(self, *comps):
 185         # This function is called a lot, so we optimize it; there are
 186         # unittests to check that we match _slow_but_correct_join(), above.
 187         path = ''
 188         sep = self.sep
 189         for comp in comps:
 190             if not comp:
 191                 continue
 192             if comp[0] == sep:
 193                 path = comp
 194                 continue
 195             if path:
 196                 path += sep
 197             path += comp
 198         if comps[-1] == '' and path:
 199             path += '/'
 200         path = path.replace(sep + sep, sep)
 201         return path
 202
 203     def listdir(self, path):
 204         root, dirs, files = list(self.walk(path))[0]
 205         return dirs + files
 206
 207     def walk(self, top):
 208         sep = self.sep
 209         if not self.isdir(top):
 210             raise OSError("%s is not a directory" % top)
 211
 212         if not top.endswith(sep):
 213             top += sep
 214
 215         dirs = []
 216         files = []
 217         for f in self.files:
 218             if self.exists(f) and f.startswith(top):
 219                 remaining = f[len(top):]
 220                 if sep in remaining:
 221                     dir = remaining[:remaining.index(sep)]
 222                     if not dir in dirs:
 223                         dirs.append(dir)
 224                 else:
 225                     files.append(remaining)
 226         return [(top[:-1], dirs, files)]
 227
 228     def mtime(self, path):
 229         if self.exists(path):
 230             return 0
 231         self._raise_not_found(path)
 232
 233     def _mktemp(self, suffix='', prefix='tmp', dir=None, **kwargs):
 234         if dir is None:
 235             dir = self.sep + '__im_tmp'
 236         curno = self.current_tmpno
 237         self.current_tmpno += 1
 238         self.last_tmpdir = self.join(dir, '%s_%u_%s' % (prefix, curno, suffix))
 239         return self.last_tmpdir
 240
 241     def mkdtemp(self, **kwargs):
 242         class TemporaryDirectory(object):
 243             def __init__(self, fs, **kwargs):
 244                 self._kwargs = kwargs
 245                 self._filesystem = fs
 246                 self._directory_path = fs._mktemp(**kwargs)
 247                 fs.maybe_make_directory(self._directory_path)
 248
 249             def __str__(self):
 250                 return self._directory_path
 251
 252             def __enter__(self):
 253                 return self._directory_path
 254
 255             def __exit__(self, type, value, traceback):
 256                 # Only self-delete if necessary.
 257
 258                 # FIXME: Should we delete non-empty directories?
 259                 if self._filesystem.exists(self._directory_path):
 260                     self._filesystem.rmtree(self._directory_path)
 261
 262         return TemporaryDirectory(fs=self, **kwargs)
 263
 264     def maybe_make_directory(self, *path):
 265         norm_path = self.normpath(self.join(*path))
 266         while norm_path and not self.isdir(norm_path):
 267             self.dirs.add(norm_path)
 268             norm_path = self.dirname(norm_path)
 269
 270     def move(self, source, destination):
 271         if self.files[source] is None:
 272             self._raise_not_found(source)
 273         self.files[destination] = self.files[source]
 274         self.written_files[destination] = self.files[destination]
 275         self.files[source] = None
 276         self.written_files[source] = None
 277
 278     def _slow_but_correct_normpath(self, path):
 279         return re.sub(re.escape(os.path.sep), self.sep, os.path.normpath(path))
 280
 281     def normpath(self, path):
 282         # This function is called a lot, so we try to optimize the common cases
 283         # instead of always calling _slow_but_correct_normpath(), above.
 284         if '..' in path or '/./' in path:
 285             # This doesn't happen very often; don't bother trying to optimize it.
 286             return self._slow_but_correct_normpath(path)
 287         if not path:
 288             return '.'
 289         if path == '/':
 290             return path
 291         if path == '/.':
 292             return '/'
 293         if path.endswith('/.'):
 294             return path[:-2]
 295         if path.endswith('/'):
 296             return path[:-1]
 297         return path
 298
 299     def open_binary_tempfile(self, suffix=''):
 300         path = self._mktemp(suffix)
 301         return (WritableBinaryFileObject(self, path), path)
 302
 303     def open_binary_file_for_reading(self, path):
 304         if self.files[path] is None:
 305             self._raise_not_found(path)
 306         return ReadableBinaryFileObject(self, path, self.files[path])
 307
 308     def read_binary_file(self, path):
 309         # Intentionally raises KeyError if we don't recognize the path.
 310         if self.files[path] is None:
 311             self._raise_not_found(path)
 312         return self.files[path]
 313
 314     def write_binary_file(self, path, contents):
 315         # FIXME: should this assert if dirname(path) doesn't exist?
 316         self.maybe_make_directory(self.dirname(path))
 317         self.files[path] = contents
 318         self.written_files[path] = contents
 319
 320     def open_text_file_for_reading(self, path):
 321         if self.files[path] is None:
 322             self._raise_not_found(path)
 323         return ReadableTextFileObject(self, path, self.files[path])
 324
 325     def open_text_file_for_writing(self, path):
 326         return WritableTextFileObject(self, path)
 327
 328     def read_text_file(self, path):
 329         return self.read_binary_file(path).decode('utf-8')
 330
 331     def write_text_file(self, path, contents):
 332         return self.write_binary_file(path, contents.encode('utf-8'))
 333
 334     def sha1(self, path):
 335         contents = self.read_binary_file(path)
 336         return hashlib.sha1(contents).hexdigest()
 337
 338     def relpath(self, path, start='.'):
 339         # Since os.path.relpath() calls os.path.normpath()
 340         # (see http://docs.python.org/library/os.path.html#os.path.abspath )
 341         # it also removes trailing slashes and converts forward and backward
 342         # slashes to the preferred slash os.sep.
 343         start = self.abspath(start)
 344         path = self.abspath(path)
 345
 346         common_root = start
 347         dot_dot = ''
 348         while not common_root == '':
 349             if path.startswith(common_root):
 350                  break
 351             common_root = self.dirname(common_root)
 352             dot_dot += '..' + self.sep
 353
 354         rel_path = path[len(common_root):]
 355
 356         if not rel_path:
 357             return '.'
 358
 359         if rel_path[0] == self.sep:
 360             # It is probably sufficient to remove just the first character
 361             # since os.path.normpath() collapses separators, but we use
 362             # lstrip() just to be sure.
 363             rel_path = rel_path.lstrip(self.sep)
 364         elif not common_root == '/':
 365             # We are in the case typified by the following example:
 366             # path = "/tmp/foobar", start = "/tmp/foo" -> rel_path = "bar"
 367             common_root = self.dirname(common_root)
 368             dot_dot += '..' + self.sep
 369             rel_path = path[len(common_root) + 1:]
 370
 371         return dot_dot + rel_path
 372
 373     def remove(self, path):
 374         if self.files[path] is None:
 375             self._raise_not_found(path)
 376         self.files[path] = None
 377         self.written_files[path] = None
 378
 379     def rmtree(self, path):
 380         path = self.normpath(path)
 381
 382         for f in self.files:
 383             # We need to add a trailing separator to path to avoid matching
 384             # cases like path='/foo/b' and f='/foo/bar/baz'.
 385             if f == path or f.startswith(path + self.sep):
 386                 self.files[f] = None
 387
 388         self.dirs = set(filter(lambda d: not (d == path or d.startswith(path + self.sep)), self.dirs))
 389
 390     def copytree(self, source, destination):
 391         source = self.normpath(source)
 392         destination = self.normpath(destination)
 393
 394         for source_file in self.files:
 395             if source_file.startswith(source):
 396                 destination_path = self.join(destination, self.relpath(source_file, source))
 397                 self.maybe_make_directory(self.dirname(destination_path))
 398                 self.files[destination_path] = self.files[source_file]
 399
 400     def split(self, path):
 401         idx = path.rfind(self.sep)
 402         if idx == -1:
 403             return ('', path)
 404         return (path[:idx], path[(idx + 1):])
 405
 406     def splitext(self, path):
 407         idx = path.rfind('.')
 408         if idx == -1:
 409             idx = len(path)
 410         return (path[0:idx], path[idx:])
 411
 412
 413 class WritableBinaryFileObject(object):
 414     def __init__(self, fs, path):
 415         self.fs = fs
 416         self.path = path
 417         self.closed = False
 418         self.fs.files[path] = ""
 419
 420     def __enter__(self):
 421         return self
 422
 423     def __exit__(self, type, value, traceback):
 424         self.close()
 425
 426     def close(self):
 427         self.closed = True
 428
 429     def write(self, str):
 430         self.fs.files[self.path] += str
 431         self.fs.written_files[self.path] = self.fs.files[self.path]
 432
 433
 434 class WritableTextFileObject(WritableBinaryFileObject):
 435     def write(self, str):
 436         WritableBinaryFileObject.write(self, str.encode('utf-8'))
 437
 438
 439 class ReadableBinaryFileObject(object):
 440     def __init__(self, fs, path, data):
 441         self.fs = fs
 442         self.path = path
 443         self.closed = False
 444         self.data = data
 445         self.offset = 0
 446
 447     def __enter__(self):
 448         return self
 449
 450     def __exit__(self, type, value, traceback):
 451         self.close()
 452
 453     def close(self):
 454         self.closed = True
 455
 456     def read(self, bytes=None):
 457         if not bytes:
 458             return self.data[self.offset:]
 459         start = self.offset
 460         self.offset += bytes
 461         return self.data[start:self.offset]
 462
 463
 464 class ReadableTextFileObject(ReadableBinaryFileObject):
 465     def __init__(self, fs, path, data):
 466         super(ReadableTextFileObject, self).__init__(fs, path, StringIO.StringIO(data.decode("utf-8")))
 467
 468     def close(self):
 469         self.data.close()
 470         super(ReadableTextFileObject, self).close()
 471
 472     def read(self, bytes=-1):
 473         return self.data.read(bytes)
 474
 475     def readline(self, length=None):
 476         return self.data.readline(length)
 477
 478     def __iter__(self):
 479         return self.data.__iter__()
 480
 481     def next(self):
 482         return self.data.next()
 483
 484     def seek(self, offset, whence=os.SEEK_SET):
 485         self.data.seek(offset, whence)