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 5 import os 6 import sys 7 8 from docs_server_utils import StringIdentity 9 from file_system import FileSystem, FileNotFoundError, StatInfo 10 from future import Future 11 from path_util import AssertIsDirectory, AssertIsValid 12 from test_util import ChromiumPath 13 14 15 def _ConvertToFilepath(path): 16 return path.replace('/', os.sep) 17 18 19 def _ConvertFromFilepath(path): 20 return path.replace(os.sep, '/') 21 22 23 def _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 31 def _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 48 def _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 67 class LocalFileSystem(FileSystem): 68 '''FileSystem implementation which fetches resources from the local 69 filesystem. 70 ''' 71 def __init__(self, base_path): 72 # Enforce POSIX path, so path validity checks pass for Windows. 73 base_path = base_path.replace(os.sep, '/') 74 AssertIsDirectory(base_path) 75 self._base_path = _ConvertToFilepath(base_path) 76 77 @staticmethod 78 def Create(*path): 79 return LocalFileSystem(ChromiumPath(*path)) 80 81 def Read(self, paths, skip_not_found=False): 82 def resolve(): 83 result = {} 84 for path in paths: 85 AssertIsValid(path) 86 full_path = os.path.join(self._base_path, 87 _ConvertToFilepath(path).lstrip(os.sep)) 88 if path == '' or path.endswith('/'): 89 result[path] = _ListDir(full_path) 90 else: 91 try: 92 result[path] = _ReadFile(full_path) 93 except FileNotFoundError: 94 if skip_not_found: 95 continue 96 return Future(exc_info=sys.exc_info()) 97 return result 98 return Future(callback=resolve) 99 100 def Refresh(self): 101 return Future(value=()) 102 103 def Stat(self, path): 104 AssertIsValid(path) 105 full_path = os.path.join(self._base_path, 106 _ConvertToFilepath(path).lstrip(os.sep)) 107 return _CreateStatInfo(full_path) 108 109 def GetIdentity(self): 110 return '@'.join((self.__class__.__name__, StringIdentity(self._base_path))) 111 112 def __repr__(self): 113 return 'LocalFileSystem(%s)' % self._base_path 114