local_file_system.py revision 5d1f7b1de12d16ceb2c938c56701a3e8bfa558f7
1# Copyright (c) 2012 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 5import os 6import sys 7 8from docs_server_utils import StringIdentity 9from file_system import FileSystem, FileNotFoundError, StatInfo 10from future import Gettable, Future 11from path_util import AssertIsDirectory, AssertIsValid 12from test_util import ChromiumPath 13 14 15def _ConvertToFilepath(path): 16 return path.replace('/', os.sep) 17 18 19def _ConvertFromFilepath(path): 20 return path.replace(os.sep, '/') 21 22 23def _ReadFile(filename): 24 try: 25 with open(filename, 'rb') as f: 26 return f.read() 27 except IOError as e: 28 raise FileNotFoundError('Read failed for %s: %s' % (filename, e)) 29 30 31def _ListDir(dir_name): 32 all_files = [] 33 try: 34 files = os.listdir(dir_name) 35 except OSError as e: 36 raise FileNotFoundError('os.listdir failed for %s: %s' % (dir_name, e)) 37 for os_path in files: 38 posix_path = _ConvertFromFilepath(os_path) 39 if os_path.startswith('.'): 40 continue 41 if os.path.isdir(os.path.join(dir_name, os_path)): 42 all_files.append(posix_path + '/') 43 else: 44 all_files.append(posix_path) 45 return all_files 46 47 48def _CreateStatInfo(path): 49 try: 50 path_mtime = os.stat(path).st_mtime 51 if os.path.isdir(path): 52 child_versions = dict((_ConvertFromFilepath(filename), 53 os.stat(os.path.join(path, filename)).st_mtime) 54 for filename in os.listdir(path)) 55 # This file system stat mimics subversion, where the stat of directories 56 # is max(file stats). That means we need to recursively check the whole 57 # file system tree :\ so approximate that by just checking this dir. 58 version = max([path_mtime] + child_versions.values()) 59 else: 60 child_versions = None 61 version = path_mtime 62 return StatInfo(version, child_versions) 63 except OSError as e: 64 raise FileNotFoundError('os.stat failed for %s: %s' % (path, e)) 65 66 67class LocalFileSystem(FileSystem): 68 '''FileSystem implementation which fetches resources from the local 69 filesystem. 70 ''' 71 def __init__(self, base_path): 72 AssertIsDirectory(base_path) 73 self._base_path = _ConvertToFilepath(base_path) 74 75 @staticmethod 76 def Create(*path): 77 return LocalFileSystem(ChromiumPath(*path)) 78 79 def Read(self, paths): 80 def resolve(): 81 result = {} 82 for path in paths: 83 AssertIsValid(path) 84 full_path = os.path.join(self._base_path, 85 _ConvertToFilepath(path).lstrip(os.sep)) 86 if path == '' or path.endswith('/'): 87 result[path] = _ListDir(full_path) 88 else: 89 result[path] = _ReadFile(full_path) 90 return result 91 return Future(delegate=Gettable(resolve)) 92 93 def Refresh(self): 94 return Future(value=()) 95 96 def Stat(self, path): 97 AssertIsValid(path) 98 full_path = os.path.join(self._base_path, 99 _ConvertToFilepath(path).lstrip(os.sep)) 100 return _CreateStatInfo(full_path) 101 102 def GetIdentity(self): 103 return '@'.join((self.__class__.__name__, StringIdentity(self._base_path))) 104 105 def __repr__(self): 106 return 'LocalFileSystem(%s)' % self._base_path 107