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.
8 from future import Future
9 from path_util import AssertIsDirectory, IsDirectory
12 class _Response(object):
13 def __init__(self, content=''):
14 self.content = content
15 self.headers = {'Content-Type': 'none'}
16 self.status_code = 200
19 class FakeUrlFetcher(object):
20 def __init__(self, base_path):
21 self._base_path = base_path
22 # Mock capabilities. Perhaps this class should be MockUrlFetcher.
25 self._async_resolve_count = 0
27 def _ReadFile(self, filename):
28 with open(os.path.join(self._base_path, filename), 'r') as f:
31 def _ListDir(self, directory):
32 # In some tests, we need to test listing a directory from the HTML returned
33 # from SVN. This reads an HTML file that has the directories HTML.
34 if not os.path.isdir(os.path.join(self._base_path, directory)):
35 return self._ReadFile(directory[:-1])
36 files = os.listdir(os.path.join(self._base_path, directory))
37 html = '<html><title>Revision: 00000</title>\n'
38 for filename in files:
39 if filename.startswith('.'):
41 if os.path.isdir(os.path.join(self._base_path, directory, filename)):
42 html += '<a>' + filename + '/</a>\n'
44 html += '<a>' + filename + '</a>\n'
48 def FetchAsync(self, url):
49 self._async_count += 1
50 url = url.rsplit('?', 1)[0]
52 self._async_resolve_count += 1
53 return self._DoFetch(url)
54 return Future(callback=resolve)
58 return self._DoFetch(url)
60 def _DoFetch(self, url):
61 url = url.rsplit('?', 1)[0]
64 result.content = self._ListDir(url)
66 result.content = self._ReadFile(url)
69 def CheckAndReset(self, sync_count=0, async_count=0, async_resolve_count=0):
70 '''Returns a tuple (success, error). Use in tests like:
71 self.assertTrue(*fetcher.CheckAndReset(...))
74 for desc, expected, actual in (
75 ('sync_count', sync_count, self._sync_count),
76 ('async_count', async_count, self._async_count),
77 ('async_resolve_count', async_resolve_count,
78 self._async_resolve_count)):
79 if actual != expected:
80 errors.append('%s: expected %s got %s' % (desc, expected, actual))
82 return (len(errors) == 0, ', '.join(errors))
89 self._async_resolve_count = 0
92 class FakeURLFSFetcher(object):
93 '''Use a file_system to resolve fake fetches. Mimics the interface of Google
97 def __init__(self, file_system, base_path):
98 AssertIsDirectory(base_path)
99 self._base_path = base_path
100 self._file_system = file_system
102 def FetchAsync(self, url, **kwargs):
103 return Future(value=self.Fetch(url))
105 def Fetch(self, url, **kwargs):
106 return _Response(self._file_system.ReadSingle(
107 posixpath.join(self._base_path, url)).Get())
109 def UpdateFS(self, file_system, base_path=None):
110 '''Replace the underlying FileSystem used to reslove URLs.
112 self._file_system = file_system
113 self._base_path = base_path or self._base_path
116 class MockURLFetcher(object):
117 def __init__(self, fetcher):
118 self._fetcher = fetcher
121 def Fetch(self, url, **kwargs):
122 self._fetch_count += 1
123 return self._fetcher.Fetch(url, **kwargs)
125 def FetchAsync(self, url, **kwargs):
126 self._fetch_async_count += 1
127 future = self._fetcher.FetchAsync(url, **kwargs)
129 self._fetch_resolve_count += 1
131 return Future(callback=resolve)
133 def CheckAndReset(self,
136 fetch_resolve_count=0):
138 for desc, expected, actual in (
139 ('fetch_count', fetch_count, self._fetch_count),
140 ('fetch_async_count', fetch_async_count, self._fetch_async_count),
141 ('fetch_resolve_count', fetch_resolve_count,
142 self._fetch_resolve_count)):
143 if actual != expected:
144 errors.append('%s: expected %s got %s' % (desc, expected, actual))
146 return (len(errors) == 0, ', '.join(errors))
151 self._fetch_count = 0
152 self._fetch_async_count = 0
153 self._fetch_resolve_count = 0