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 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    # 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