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 not self.exists(source):
 272             self._raise_not_found(source)
 273         if self.isfile(source):
 274             self.files[destination] = self.files[source]
 275             self.written_files[destination] = self.files[destination]
 276             self.files[source] = None
 277             self.written_files[source] = None
 278             return
 279         self.copytree(source, destination)
 280         self.rmtree(source)
 281
 282     def _slow_but_correct_normpath(self, path):
 283         return re.sub(re.escape(os.path.sep), self.sep, os.path.normpath(path))
 284
 285     def normpath(self, path):
 286         # This function is called a lot, so we try to optimize the common cases
 287         # instead of always calling _slow_but_correct_normpath(), above.
 288         if '..' in path or '/./' in path:
 289             # This doesn't happen very often; don't bother trying to optimize it.
 290             return self._slow_but_correct_normpath(path)
 291         if not path:
 292             return '.'
 293         if path == '/':
 294             return path
 295         if path == '/.':
 296             return '/'
 297         if path.endswith('/.'):
 298             return path[:-2]
 299         if path.endswith('/'):
 300             return path[:-1]
 301         return path
 302
 303     def open_binary_tempfile(self, suffix=''):
 304         path = self._mktemp(suffix)
 305         return (WritableBinaryFileObject(self, path), path)
 306
 307     def open_binary_file_for_reading(self, path):
 308         if self.files[path] is None:
 309             self._raise_not_found(path)
 310         return ReadableBinaryFileObject(self, path, self.files[path])
 311
 312     def read_binary_file(self, path):
 313         # Intentionally raises KeyError if we don't recognize the path.
 314         if self.files[path] is None:
 315             self._raise_not_found(path)
 316         return self.files[path]
 317
 318     def write_binary_file(self, path, contents):
 319         # FIXME: should this assert if dirname(path) doesn't exist?
 320         self.maybe_make_directory(self.dirname(path))
 321         self.files[path] = contents
 322         self.written_files[path] = contents
 323
 324     def open_text_file_for_reading(self, path):
 325         if self.files[path] is None:
 326             self._raise_not_found(path)
 327         return ReadableTextFileObject(self, path, self.files[path])
 328
 329     def open_text_file_for_writing(self, path):
 330         return WritableTextFileObject(self, path)
 331
 332     def read_text_file(self, path):
 333         return self.read_binary_file(path).decode('utf-8')
 334
 335     def write_text_file(self, path, contents):
 336         return self.write_binary_file(path, contents.encode('utf-8'))
 337
 338     def sha1(self, path):
 339         contents = self.read_binary_file(path)
 340         return hashlib.sha1(contents).hexdigest()
 341
 342     def relpath(self, path, start='.'):
 343         # Since os.path.relpath() calls os.path.normpath()
 344         # (see http://docs.python.org/library/os.path.html#os.path.abspath )
 345         # it also removes trailing slashes and converts forward and backward
 346         # slashes to the preferred slash os.sep.
 347         start = self.abspath(start)
 348         path = self.abspath(path)
 349
 350         common_root = start
 351         dot_dot = ''
 352         while not common_root == '':
 353             if path.startswith(common_root):
 354                  break
 355             common_root = self.dirname(common_root)
 356             dot_dot += '..' + self.sep
 357
 358         rel_path = path[len(common_root):]
 359
 360         if not rel_path:
 361             return '.'
 362
 363         if rel_path[0] == self.sep:
 364             # It is probably sufficient to remove just the first character
 365             # since os.path.normpath() collapses separators, but we use
 366             # lstrip() just to be sure.
 367             rel_path = rel_path.lstrip(self.sep)
 368         elif not common_root == '/':
 369             # We are in the case typified by the following example:
 370             # path = "/tmp/foobar", start = "/tmp/foo" -> rel_path = "bar"
 371             common_root = self.dirname(common_root)
 372             dot_dot += '..' + self.sep
 373             rel_path = path[len(common_root) + 1:]
 374
 375         return dot_dot + rel_path
 376
 377     def remove(self, path):
 378         if self.files[path] is None:
 379             self._raise_not_found(path)
 380         self.files[path] = None
 381         self.written_files[path] = None
 382
 383     def rmtree(self, path):
 384         path = self.normpath(path)
 385
 386         for f in self.files:
 387             # We need to add a trailing separator to path to avoid matching
 388             # cases like path='/foo/b' and f='/foo/bar/baz'.
 389             if f == path or f.startswith(path + self.sep):
 390                 self.files[f] = None
 391
 392         self.dirs = set(filter(lambda d: not (d == path or d.startswith(path + self.sep)), self.dirs))
 393
 394     def copytree(self, source, destination):
 395         source = self.normpath(source)
 396         destination = self.normpath(destination)
 397
 398         for source_file in list(self.files):
 399             if source_file.startswith(source):
 400                 destination_path = self.join(destination, self.relpath(source_file, source))
 401                 self.maybe_make_directory(self.dirname(destination_path))
 402                 self.files[destination_path] = self.files[source_file]
 403
 404     def split(self, path):
 405         idx = path.rfind(self.sep)
 406         if idx == -1:
 407             return ('', path)
 408         return (path[:idx], path[(idx + 1):])
 409
 410     def splitext(self, path):
 411         idx = path.rfind('.')
 412         if idx == -1:
 413             idx = len(path)
 414         return (path[0:idx], path[idx:])
 415
 416
 417 class WritableBinaryFileObject(object):
 418     def __init__(self, fs, path):
 419         self.fs = fs
 420         self.path = path
 421         self.closed = False
 422         self.fs.files[path] = ""
 423
 424     def __enter__(self):
 425         return self
 426
 427     def __exit__(self, type, value, traceback):
 428         self.close()
 429
 430     def close(self):
 431         self.closed = True
 432
 433     def write(self, str):
 434         self.fs.files[self.path] += str
 435         self.fs.written_files[self.path] = self.fs.files[self.path]
 436
 437
 438 class WritableTextFileObject(WritableBinaryFileObject):
 439     def write(self, str):
 440         WritableBinaryFileObject.write(self, str.encode('utf-8'))
 441
 442
 443 class ReadableBinaryFileObject(object):
 444     def __init__(self, fs, path, data):
 445         self.fs = fs
 446         self.path = path
 447         self.closed = False
 448         self.data = data
 449         self.offset = 0
 450
 451     def __enter__(self):
 452         return self
 453
 454     def __exit__(self, type, value, traceback):
 455         self.close()
 456
 457     def close(self):
 458         self.closed = True
 459
 460     def read(self, bytes=None):
 461         if not bytes:
 462             return self.data[self.offset:]
 463         start = self.offset
 464         self.offset += bytes
 465         return self.data[start:self.offset]
 466
 467
 468 class ReadableTextFileObject(ReadableBinaryFileObject):
 469     def __init__(self, fs, path, data):
 470         super(ReadableTextFileObject, self).__init__(fs, path, StringIO.StringIO(data.decode("utf-8")))
 471
 472     def close(self):
 473         self.data.close()
 474         super(ReadableTextFileObject, self).close()
 475
 476     def read(self, bytes=-1):
 477         return self.data.read(bytes)
 478
 479     def readline(self, length=None):
 480         return self.data.readline(length)
 481
 482     def __iter__(self):
 483         return self.data.__iter__()
 484
 485     def next(self):
 486         return self.data.next()
 487
 488     def seek(self, offset, whence=os.SEEK_SET):
 489         self.data.seek(offset, whence)