chrome/common/extensions/docs/server2/mock_file_system.py

   1 # Copyright 2013 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 import posixpath
   6
   7 from file_system import FileSystem, FileNotFoundError
   8 from future import Future
   9 from test_file_system import _List, _StatTracker, TestFileSystem
  10 from path_util import IsDirectory
  11
  12
  13 class MockFileSystem(FileSystem):
  14   '''Wraps FileSystems to add a selection of mock behaviour:
  15   - asserting how often Stat/Read calls are being made to it.
  16   - primitive changes/versioning via applying object "diffs", mapping paths to
  17     new content (similar to how TestFileSystem works).
  18   '''
  19   def __init__(self, file_system):
  20     self._file_system = file_system
  21     # Updates are stored as TestFileSystems because it already implements a
  22     # bunch of logic to intepret paths into dictionaries.
  23     self._updates = []
  24     self._stat_tracker = _StatTracker()
  25     self._read_count = 0
  26     self._read_resolve_count = 0
  27     self._stat_count = 0
  28     self._version = None
  29
  30   @staticmethod
  31   def Create(file_system, updates):
  32     mock_file_system = MockFileSystem(file_system)
  33     for update in updates:
  34       mock_file_system.Update(update)
  35     return mock_file_system
  36
  37   #
  38   # FileSystem implementation.
  39   #
  40
  41   def Read(self, paths, skip_not_found=False):
  42     '''Reads |paths| from |_file_system|, then applies the most recent update
  43     from |_updates|, if any.
  44     '''
  45     self._read_count += 1
  46     def next(result):
  47       self._read_resolve_count += 1
  48       for path in result.iterkeys():
  49         update = self._GetMostRecentUpdate(path)
  50         if update is not None:
  51           result[path] = update
  52       return result
  53     return self._file_system.Read(paths,
  54                                   skip_not_found=skip_not_found).Then(next)
  55
  56   def Refresh(self):
  57     return self._file_system.Refresh()
  58
  59   def _GetMostRecentUpdate(self, path):
  60     '''Returns the latest update for the file at |path|, or None if |path|
  61     has never been updated.
  62     '''
  63     for update in reversed(self._updates):
  64       try:
  65         return update.ReadSingle(path).Get()
  66       except FileNotFoundError:
  67         pass
  68     return None
  69
  70   def Stat(self, path):
  71     self._stat_count += 1
  72
  73     # This only supports numeric stat values since we need to add to it.  In
  74     # reality the logic here could just be to randomly mutate the stat values
  75     # every time there's an Update but that's less meaningful for testing.
  76     def stradd(a, b):
  77       return str(int(a) + b)
  78
  79     stat = self._file_system.Stat(path)
  80     stat.version = stradd(stat.version, self._stat_tracker.GetVersion(path))
  81     if stat.child_versions:
  82       for child_path, child_version in stat.child_versions.iteritems():
  83         stat.child_versions[child_path] = stradd(
  84             stat.child_versions[child_path],
  85             self._stat_tracker.GetVersion(posixpath.join(path, child_path)))
  86
  87     return stat
  88
  89   def GetCommitID(self):
  90     return Future(value=str(self._stat_tracker.GetVersion('')))
  91
  92   def GetPreviousCommitID(self):
  93     return Future(value=str(self._stat_tracker.GetVersion('') - 1))
  94
  95   def GetIdentity(self):
  96     return self._file_system.GetIdentity()
  97
  98   def GetVersion(self):
  99     return self._version
 100
 101   def __str__(self):
 102     return repr(self)
 103
 104   def __repr__(self):
 105     return 'MockFileSystem(read_count=%s, stat_count=%s, updates=%s)' % (
 106         self._read_count, self._stat_count, len(self._updates))
 107
 108   #
 109   # Testing methods.
 110   #
 111
 112   def GetStatCount(self):
 113     return self._stat_count
 114
 115   def CheckAndReset(self, stat_count=0, read_count=0, read_resolve_count=0):
 116     '''Returns a tuple (success, error). Use in tests like:
 117     self.assertTrue(*object_store.CheckAndReset(...))
 118     '''
 119     errors = []
 120     for desc, expected, actual in (
 121         ('read_count', read_count, self._read_count),
 122         ('read_resolve_count', read_resolve_count, self._read_resolve_count),
 123         ('stat_count', stat_count, self._stat_count)):
 124       if actual != expected:
 125         errors.append('%s: expected %s got %s' % (desc, expected, actual))
 126     try:
 127       return (len(errors) == 0, ', '.join(errors))
 128     finally:
 129       self.Reset()
 130
 131   def Reset(self):
 132     self._read_count = 0
 133     self._read_resolve_count = 0
 134     self._stat_count = 0
 135
 136   def Update(self, update):
 137     self._updates.append(TestFileSystem(update))
 138     for path in _List(update).iterkeys():
 139       # Any files (not directories) which changed are now at the version
 140       # derived from |_updates|.
 141       if not IsDirectory(path):
 142         self._stat_tracker.SetVersion(path, len(self._updates))
 143
 144   def SetVersion(self, version):
 145     '''Override the reported FileSystem version (default None) for testing.'''
 146     self._version = version