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 from file_system import FileSystem, FileNotFoundError
   6 from future import Gettable, Future
   7 from test_file_system import TestFileSystem
   8
   9 class MockFileSystem(FileSystem):
  10   '''Wraps FileSystems to add a selection of mock behaviour:
  11
  12   - asserting how often Stat/Read calls are being made to it.
  13   - primitive changes/versioning via applying object "diffs", mapping paths to
  14     new content (similar to how TestFileSystem works).
  15   '''
  16   def __init__(self, file_system):
  17     self._file_system = file_system
  18     # Updates are modelled are stored as TestFileSystems because they've
  19     # implemented a bunch of logic to interpret paths into dictionaries.
  20     self._updates = []
  21     self._read_count = 0
  22     self._read_resolve_count = 0
  23     self._stat_count = 0
  24
  25   @staticmethod
  26   def Create(file_system, updates):
  27     mock_file_system = MockFileSystem(file_system)
  28     for update in updates:
  29       mock_file_system.Update(update)
  30     return mock_file_system
  31
  32   #
  33   # FileSystem implementation.
  34   #
  35
  36   def Read(self, paths):
  37     '''Reads |paths| from |_file_system|, then applies the most recent update
  38     from |_updates|, if any.
  39     '''
  40     self._read_count += 1
  41     future_result = self._file_system.Read(paths)
  42     def resolve():
  43       self._read_resolve_count += 1
  44       result = future_result.Get()
  45       for path in result.iterkeys():
  46         _, update = self._GetMostRecentUpdate(path)
  47         if update is not None:
  48           result[path] = update
  49       return result
  50     return Future(delegate=Gettable(resolve))
  51
  52   def Refresh(self):
  53     return self._file_system.Refresh()
  54
  55   def _GetMostRecentUpdate(self, path):
  56     for revision, update in reversed(list(enumerate(self._updates))):
  57       try:
  58         return (revision + 1, update.ReadSingle(path).Get())
  59       except FileNotFoundError:
  60         pass
  61     return (0, None)
  62
  63   def Stat(self, path):
  64     self._stat_count += 1
  65     return self._StatImpl(path)
  66
  67   def _StatImpl(self, path):
  68     result = self._file_system.Stat(path)
  69     result.version = self._UpdateStat(result.version, path)
  70     child_versions = result.child_versions
  71     if child_versions is not None:
  72       for child_path in child_versions.iterkeys():
  73         child_versions[child_path] = self._UpdateStat(
  74             child_versions[child_path],
  75             '%s%s' % (path, child_path))
  76     return result
  77
  78   def _UpdateStat(self, version, path):
  79     if not path.endswith('/'):
  80       return str(int(version) + self._GetMostRecentUpdate(path)[0])
  81     # Bleh, it's a directory, need to recursively search all the children.
  82     child_paths = self._file_system.ReadSingle(path).Get()
  83     if not child_paths:
  84       return version
  85     return str(max([int(version)] +
  86                    [int(self._StatImpl('%s%s' % (path, child_path)).version)
  87                     for child_path in child_paths]))
  88
  89   def GetIdentity(self):
  90     return self._file_system.GetIdentity()
  91
  92   def __str__(self):
  93     return repr(self)
  94
  95   def __repr__(self):
  96     return 'MockFileSystem(read_count=%s, stat_count=%s, updates=%s)' % (
  97         self._read_count, self._stat_count, len(self._updates))
  98
  99   #
 100   # Testing methods.
 101   #
 102
 103   def GetStatCount(self):
 104     return self._stat_count
 105
 106   def CheckAndReset(self, stat_count=0, read_count=0, read_resolve_count=0):
 107     '''Returns a tuple (success, error). Use in tests like:
 108     self.assertTrue(*object_store.CheckAndReset(...))
 109     '''
 110     errors = []
 111     for desc, expected, actual in (
 112         ('read_count', read_count, self._read_count),
 113         ('read_resolve_count', read_resolve_count, self._read_resolve_count),
 114         ('stat_count', stat_count, self._stat_count)):
 115       if actual != expected:
 116         errors.append('%s: expected %s got %s' % (desc, expected, actual))
 117     try:
 118       return (len(errors) == 0, ', '.join(errors))
 119     finally:
 120       self.Reset()
 121
 122   def Reset(self):
 123     self._read_count = 0
 124     self._read_resolve_count = 0
 125     self._stat_count = 0
 126
 127   def Update(self, update):
 128     self._updates.append(TestFileSystem(update))